diff --git a/roles/translator/src/lib/channel_manager/mod.rs b/roles/translator/src/lib/channel_manager/mod.rs new file mode 100644 index 0000000000..117d7392f0 --- /dev/null +++ b/roles/translator/src/lib/channel_manager/mod.rs @@ -0,0 +1,575 @@ +#![allow(warnings)] +//! What should be the role of channel manager? +//! +//! 1. It should assign extranonce to new connection. +//! 2. It can coordinate with upstream module to open connection in case of non-aggregation. +//! 3. It should perform share validation and send correct response to corresponding downstream. +//! 4. It should be responsible for difficulty management for each sv1 channel. +//! 5. It should harbour the jobs received from upstream, to perform validation correctly. + +/// We gonna be having two flows, one for aggregation and another for non-aggregation +/// In case of aggregation, the whole tproxy flow gonna start from the upstream submodule +/// where the upstream gonna connect to Pool/JDC by itself at the beginning of the setup. +/// In case of non-aggregation, the whole tproxy flow gonna start from downstream, where +/// once a downstream connects we gonna open a extended mining channel with the upstream. +use std::collections::{HashMap, HashSet}; + +use binary_sv2::u256_from_int; +use roles_logic_sv2::{ + channels::server::{jobs::extended::ExtendedJob, share_accounting::ShareAccounting}, + mining_sv2::{ExtendedExtranonce, Extranonce, NewExtendedMiningJob, SetNewPrevHash, Target}, + utils::{bytes_to_hex, merkle_root_from_path, target_to_difficulty, u256_to_block_hash, Id}, +}; +use stratum_common::bitcoin::{ + blockdata::block::{Header, Version}, + hashes::sha256d::Hash, + transaction::TxOut, + CompactTarget, Target as BitcoinTarget, +}; +use tracing::{debug, info}; +use v1::utils::HexU32Be; + +use crate::{ + config::UpstreamDifficultyConfig, downstream_sv1::SubmitShareWithChannelId, + utils::proxy_extranonce1_len, +}; + +#[derive(PartialEq, Hash, Eq, Clone, Debug, Copy)] +pub struct Sv1ChannelId(u32); + +/// Sv1 channel representation +pub struct Sv1Channel { + // Channel id of the connection + channel_id: Sv1ChannelId, + // User identity + user_identity: String, + // Extranonce prefix allocated for the connection + extranonce_prefix: Vec, + // Rollable extranonce size for the connection + rollable_extranonce_size: u16, + /// Version rolling mask bits + version_rolling_mask: Option, + /// Minimum version rolling mask bits size + version_rolling_min_bit: Option, +} + +impl Sv1Channel { + fn new( + channel_id: Sv1ChannelId, + user_identity: String, + extranonce_prefix: Vec, + rollable_extranonce_size: u16, + ) -> Self { + Self { + channel_id, + user_identity, + extranonce_prefix, + rollable_extranonce_size, + version_rolling_mask: None, + version_rolling_min_bit: None, + } + } +} + +#[derive(Debug)] +pub struct UpstreamChannelManager { + pub channel_ids: HashSet, + pub request_id_to_channel_id: HashMap, + pub upstream_manager: HashMap, + pub aggregate: bool, + pub min_extranonce_size: u16, + pub bootstrap_nominal_hashrate: f32, + pub update_interval: u32, + pub shares_per_minute: f32, +} + +#[derive(Debug)] +pub struct UpstreamChannel { + pub downstream_manager: ChannelManager, + pub last_sent_hashrate: f32, + pub upstream_difficulty: UpstreamDifficultyConfig, + pub target: Target, +} + +impl UpstreamChannel { + pub fn new( + downstream_manager: ChannelManager, + last_sent_hashrate: f32, + upstream_difficulty: UpstreamDifficultyConfig, + target: Target, + ) -> Self { + Self { + downstream_manager, + last_sent_hashrate, + upstream_difficulty, + target, + } + } +} + +impl UpstreamChannelManager { + pub fn new( + min_extranonce_size: u16, + bootstrap_nominal_hashrate: f32, + update_interval: u32, + shares_per_minute: f32, + ) -> Self { + Self { + channel_ids: HashSet::new(), + request_id_to_channel_id: HashMap::new(), + upstream_manager: HashMap::new(), + aggregate: true, + min_extranonce_size, + bootstrap_nominal_hashrate, + update_interval, + shares_per_minute, + } + } + + pub fn remove(&mut self, id: u32) { + self.channel_ids.remove(&id); + // todo: Improve this later + self.upstream_manager.remove(&id); + } + + pub fn downstream_difficulty_hashrate( + &self, + channel_id: u32, + connection_id: Sv1ChannelId, + ) -> Option { + if let Some(upstream_channel) = self.upstream_manager.get(&channel_id) { + if let Some(difficulty_manager) = upstream_channel + .downstream_manager + .difficulty_config + .get(&connection_id) + { + return Some(difficulty_manager.min_individual_miner_hashrate.clone()); + } + } + None + } + + pub fn downstream_difficulty_target( + &self, + channel_id: u32, + connection_id: Sv1ChannelId, + ) -> Option { + if let Some(upstream_channel) = self.upstream_manager.get(&channel_id) { + if let Some(difficulty_manager) = upstream_channel + .downstream_manager + .difficulty_config + .get(&connection_id) + { + return Some(difficulty_manager.target.clone()); + } + } + None + } + + pub fn downstream_difficulty_submits_since_last_update( + &self, + channel_id: u32, + connection_id: Sv1ChannelId, + ) -> Option { + if let Some(upstream_channel) = self.upstream_manager.get(&channel_id) { + if let Some(difficulty_manager) = upstream_channel + .downstream_manager + .difficulty_config + .get(&connection_id) + { + return Some(difficulty_manager.submits_since_last_update.clone()); + } + } + None + } + + pub fn downstream_difficulty_timestamp_of_last_update( + &self, + channel_id: u32, + connection_id: Sv1ChannelId, + ) -> Option { + if let Some(upstream_channel) = self.upstream_manager.get(&channel_id) { + if let Some(difficulty_manager) = upstream_channel + .downstream_manager + .difficulty_config + .get(&connection_id) + { + return Some(difficulty_manager.timestamp_of_last_update.clone()); + } + } + None + } + + pub fn set_downstream_difficulty_hashrate( + &mut self, + channel_id: u32, + connection_id: Sv1ChannelId, + hashrate: f32, + ) { + if let Some(upstream_channel) = self.upstream_manager.get_mut(&channel_id) { + if let Some(difficulty_manager) = upstream_channel + .downstream_manager + .difficulty_config + .get_mut(&connection_id) + { + difficulty_manager.min_individual_miner_hashrate = hashrate; + } + } + } + + pub fn set_downstream_difficulty_target( + &mut self, + channel_id: u32, + connection_id: Sv1ChannelId, + target: Target, + ) { + if let Some(upstream_channel) = self.upstream_manager.get_mut(&channel_id) { + if let Some(difficulty_manager) = upstream_channel + .downstream_manager + .difficulty_config + .get_mut(&connection_id) + { + difficulty_manager.target = target; + } + } + } + + pub fn set_downstream_difficulty_submits_since_last_update( + &mut self, + channel_id: u32, + connection_id: Sv1ChannelId, + last_update: u32, + ) { + if let Some(upstream_channel) = self.upstream_manager.get_mut(&channel_id) { + if let Some(difficulty_manager) = upstream_channel + .downstream_manager + .difficulty_config + .get_mut(&connection_id) + { + difficulty_manager.submits_since_last_update = last_update; + } + } + } + + pub fn set_downstream_difficulty_timestamp_of_last_update( + &mut self, + channel_id: u32, + connection_id: Sv1ChannelId, + timestamp_since_last_update: u64, + ) { + if let Some(upstream_channel) = self.upstream_manager.get_mut(&channel_id) { + if let Some(difficulty_manager) = upstream_channel + .downstream_manager + .difficulty_config + .get_mut(&connection_id) + { + difficulty_manager.timestamp_of_last_update = timestamp_since_last_update; + } + } + } +} + +// Just struct this for non-aggregation case first. +#[derive(Debug)] +pub struct ChannelManager { + // Channel extranonce distributor. + pub extended_extranonce_factory: ExtendedExtranonce, + // Share account. + pub share_accounting: HashMap, + // expected share per minute from config. + pub expected_share_per_minute: f32, + // Difficulty config per connection + pub difficulty_config: HashMap, + // ID generator + pub downstream_id_factory: Id, + // Prevhash + pub prev_block_hash: Option>, + // future jobs are indexed with job_id (u32) + pub future_jobs: HashMap>, + // Currently active job shared by upstream + pub active_job: Option>, + // past jobs are indexed with job_id (u32) + pub past_jobs: HashMap>, + // stale jobs are indexed with job_id (u32) + pub stale_jobs: HashMap>, + // Channel id + pub channel_id: u32, +} + +#[derive(Debug, Clone)] +pub struct DownstreamDifficultyConfig { + pub min_individual_miner_hashrate: f32, + pub target: Target, + pub submits_since_last_update: u32, + pub timestamp_of_last_update: u64, +} + +impl DownstreamDifficultyConfig { + fn new() -> Self { + Self { + min_individual_miner_hashrate: 10_000_000_000_000.0, + submits_since_last_update: 0, + timestamp_of_last_update: 0, + target: u256_from_int(u64::MAX).into(), + } + } +} + +impl ChannelManager { + pub fn new( + extranonce_prefix: Extranonce, + extranonce_prefix_len: usize, + extranonce_size: usize, + min_extranonce_size: usize, + expected_share_per_minute: f32, + channel_id: u32, + ) -> Self { + let tproxy_len = proxy_extranonce1_len(extranonce_size, min_extranonce_size); + let range_0 = 0..extranonce_prefix_len; + let range_1 = extranonce_prefix_len..extranonce_prefix_len + tproxy_len; + let range_2 = extranonce_prefix_len + tproxy_len..extranonce_prefix_len + extranonce_size; + let extended_extranonce_factory = ExtendedExtranonce::from_upstream_extranonce( + extranonce_prefix, + range_0, + range_1, + range_2, + ) + .expect("Something went wrong extranonce factory"); + + Self { + extended_extranonce_factory, + share_accounting: HashMap::new(), + expected_share_per_minute, + difficulty_config: HashMap::new(), + downstream_id_factory: Id::new(), + prev_block_hash: None, + future_jobs: HashMap::new(), + active_job: None, + past_jobs: HashMap::new(), + stale_jobs: HashMap::new(), + channel_id, + } + } + + /// What I need to do: + /// 1. I should generate an Id to it. + /// 2. I should assign a extranonce field for new downstream + /// 3. I should add an entry in share_accounter + /// 4. I should add an entry in difficulty_config + pub fn on_new_downstream_connection( + &mut self, + user_identity: String, + ) -> (u32, Sv1ChannelId, Vec, usize) { + let new_downstream_id = Sv1ChannelId(self.downstream_id_factory.next()); + let max_extranonce2_len = self.extended_extranonce_factory.get_range2_len() as usize; + let new_extranonce = self + .extended_extranonce_factory + .next_prefix_extended(max_extranonce2_len) + .expect("Should have generated the extranonce prefix"); + let sv1_object = Sv1Channel::new( + new_downstream_id.clone(), + user_identity, + new_extranonce.clone().to_vec(), + max_extranonce2_len as u16, + ); + self.share_accounting + .insert(new_downstream_id.clone(), ShareAccounting::new(0)); + self.difficulty_config + .insert(new_downstream_id.clone(), DownstreamDifficultyConfig::new()); + ( + self.channel_id, + new_downstream_id, + new_extranonce.to_vec(), + max_extranonce2_len, + ) + } + + /// validated whether share is acceptable or not + /// Then share the result to downstream and upstream (if accepted) + /// Check against active and past jobs. + pub fn on_submit_share(&mut self, share: SubmitShareWithChannelId) -> bool { + info!("Got submit share message in channel manager"); + let job_id = share.share.job_id.parse::().unwrap(); + match self.active_job.clone() { + Some(active_job) => { + if job_id == active_job.job_id { + return self.validate_share(share, Some(active_job)); + } + + if self.past_jobs.contains_key(&job_id) { + return self.validate_share(share, self.past_jobs.get(&job_id).cloned()); + } + + return false; + } + None => return false, + } + } + + pub fn validate_share( + &mut self, + share: SubmitShareWithChannelId, + job: Option>, + ) -> bool { + info!("Got share {share:?}, for this job {job:?} in channel manager"); + + let job_id = share.share.job_id; + + if let Some(active_job) = job { + let extranonce_prefix = share.extranonce; + let mut full_extranonce = vec![]; + full_extranonce.extend(extranonce_prefix); + full_extranonce.extend(share.share.extra_nonce2.as_ref()); + + let merkle_root: [u8; 32] = merkle_root_from_path( + active_job.coinbase_tx_prefix.inner_as_ref(), + active_job.coinbase_tx_suffix.inner_as_ref(), + &full_extranonce, + &active_job.merkle_path.inner_as_ref(), + ) + .unwrap() + .try_into() + .expect("merkle root must be 32 bytes"); + + if let Some(prev_hash) = self.prev_block_hash.as_ref() { + let prev_block_hash = prev_hash.prev_hash.clone(); + let nbits = CompactTarget::from_consensus(prev_hash.nbits); + + let request_version = share + .share + .version_bits + .clone() + .map(|vb| vb.0) + .unwrap_or(active_job.version); + + let mask = share + .version_rolling_mask + .unwrap_or(HexU32Be(0x1FFFE000_u32)) + .0; + + let version = (active_job.version & !mask) | (request_version & mask); + + // create the header for validation + let header = Header { + version: Version::from_consensus(version as i32), + prev_blockhash: u256_to_block_hash(prev_block_hash.clone()), + merkle_root: (*Hash::from_bytes_ref(&merkle_root)).into(), + time: share.share.time.0, + bits: nbits, + nonce: share.share.nonce.0, + }; + + // convert the header hash to a target type for easy comparison + let hash = header.block_hash(); + let raw_hash: [u8; 32] = *hash.to_raw_hash().as_ref(); + let hash_as_target: Target = raw_hash.into(); + let hash_as_diff = target_to_difficulty(hash_as_target.clone()); + + let network_target = BitcoinTarget::from_compact(nbits); + + // print hash_as_target and self.target as human readable hex + let hash_as_u256: binary_sv2::U256 = hash_as_target.clone().into(); + let mut hash_bytes = hash_as_u256.to_vec(); + hash_bytes.reverse(); // Convert to big-endian for display + + let difficulty_config = self + .difficulty_config + .get(&Sv1ChannelId(active_job.channel_id)); + + if let Some(difficulty) = difficulty_config { + let target = difficulty.target.clone(); + let target_u256: binary_sv2::U256 = target.clone().into(); + let mut target_bytes = target_u256.to_vec(); + target_bytes.reverse(); + + debug!( + "share validation \nshare:\t\t{}\nchannel target:\t{}\nnetwork target:\t{}", + bytes_to_hex(&hash_bytes), + bytes_to_hex(&target_bytes), + format!("{:x}", network_target) + ); + + if hash_as_target <= target { + let share_accounting = self + .share_accounting + .get_mut(&Sv1ChannelId(active_job.channel_id)); + if let Some(share_accounting) = share_accounting { + if share_accounting.is_share_seen(hash.to_raw_hash()) { + return false; + } + share_accounting.update_share_accounting( + target_to_difficulty(target.clone()) as u64, + share.share.time.0, + hash.to_raw_hash(), + ); + share_accounting.update_best_diff(hash_as_diff); + let last_sequence_number = + share_accounting.get_last_share_sequence_number(); + let new_submits_accepted_count = share_accounting.get_shares_accepted(); + let new_shares_sum = share_accounting.get_share_work_sum(); + + return true; + } + } + } + } + } + + return false; + } + + pub fn on_new_prev_hash(&mut self, set_new_prevhash: SetNewPrevHash<'static>) { + info!("Received new previous block hash in channel manager: {set_new_prevhash:?}"); + let job_id = set_new_prevhash.job_id; + self.active_job = None; + if self.future_jobs.contains_key(&job_id) { + self.active_job = self.future_jobs.get(&job_id).cloned(); + } + self.prev_block_hash = Some(set_new_prevhash); + self.future_jobs.clear(); + self.past_jobs.clear(); + self.stale_jobs.clear(); + } + + pub fn on_new_extended_job(&mut self, extended_job: NewExtendedMiningJob<'static>) { + info!("Received extended mining job in channel manager: {extended_job:?}"); + if extended_job.is_future() { + self.future_jobs.insert(extended_job.job_id, extended_job); + return; + } + if self.active_job.is_none() { + self.active_job = Some(extended_job); + return; + } + + let past_active_job = self.active_job.take().expect("Active job should be active"); + self.active_job = Some(extended_job); + self.past_jobs + .insert(past_active_job.job_id, past_active_job); + } + + pub fn active_job(&self) -> Option> { + self.active_job.clone() + } + + pub fn current_prev_block_hash(&self) -> Option> { + self.prev_block_hash.clone() + } + + pub fn get_job(&self, job_id: u32) -> Option> { + if let Some(active_job) = self.active_job.clone() { + if active_job.job_id == job_id { + return Some(active_job); + } + } + + let job = self.past_jobs.get(&job_id); + if let Some(job) = job { + if job.job_id == job_id { + return Some(job.to_owned()); + } + } + + None + } +} diff --git a/roles/translator/src/lib/downstream_sv1/diff_management.rs b/roles/translator/src/lib/downstream_sv1/diff_management.rs index dbded1e882..a76c9e010e 100644 --- a/roles/translator/src/lib/downstream_sv1/diff_management.rs +++ b/roles/translator/src/lib/downstream_sv1/diff_management.rs @@ -32,29 +32,53 @@ impl Downstream { self_: Arc>, init_target: &[u8], ) -> ProxyResult<'static, ()> { - let (connection_id, upstream_difficulty_config, miner_hashrate) = self_.safe_lock(|d| { + let (channel_id, connection_id, miner_hashrate) = self_.safe_lock(|d| { let timestamp_secs = std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) .expect("time went backwards") .as_secs(); - d.difficulty_mgmt.timestamp_of_last_update = timestamp_secs; - d.difficulty_mgmt.submits_since_last_update = 0; - ( - d.connection_id, - d.upstream_difficulty_config.clone(), - d.difficulty_mgmt.min_individual_miner_hashrate, - ) + let min_individual_miner_hashrate = d + .upstream_channel_manager + .super_safe_lock(|upstream_channel_manager| { + upstream_channel_manager.set_downstream_difficulty_timestamp_of_last_update( + d.channel_id, + d.connection_id, + timestamp_secs, + ); + upstream_channel_manager.set_downstream_difficulty_submits_since_last_update( + d.channel_id, + d.connection_id, + 0, + ); + upstream_channel_manager + .downstream_difficulty_hashrate(d.channel_id, d.connection_id) + }) + .unwrap(); + (d.channel_id, d.connection_id, min_individual_miner_hashrate) })?; - // add new connection hashrate to channel hashrate - upstream_difficulty_config.safe_lock(|u| { - u.channel_nominal_hashrate += miner_hashrate; + + self_.safe_lock(|downstream| { + _ = downstream + .upstream_channel_manager + .safe_lock(|upstream_channel_manager| { + let upstream_channel = upstream_channel_manager + .upstream_manager + .get_mut(&downstream.channel_id); + if let Some(upstream_channel) = upstream_channel { + upstream_channel + .upstream_difficulty + .channel_nominal_hashrate += miner_hashrate; + } + }); })?; + // update downstream target with bridge let init_target = binary_sv2::U256::try_from(init_target.to_vec())?; Self::send_message_upstream( self_, DownstreamMessages::SetDownstreamTarget(SetDownstreamTarget { - channel_id: connection_id, + connection_id, + channel_id, new_target: init_target.into(), }), ) @@ -70,18 +94,37 @@ impl Downstream { /// the channel to the upstream server. #[allow(clippy::result_large_err)] pub fn remove_miner_hashrate_from_channel(self_: Arc>) -> ProxyResult<'static, ()> { - self_.safe_lock(|d| { - d.upstream_difficulty_config - .safe_lock(|u| { - let hashrate_to_subtract = d.difficulty_mgmt.min_individual_miner_hashrate; - if u.channel_nominal_hashrate >= hashrate_to_subtract { - u.channel_nominal_hashrate -= hashrate_to_subtract; - } else { - u.channel_nominal_hashrate = 0.0; + self_.safe_lock(|downstream| { + _ = downstream + .upstream_channel_manager + .safe_lock(|upstream_channel_manager| { + let hashrate_to_substract = upstream_channel_manager + .downstream_difficulty_hashrate( + downstream.channel_id, + downstream.connection_id, + ); + let upstream_channel = upstream_channel_manager + .upstream_manager + .get_mut(&downstream.channel_id); + if let Some(upstream_channel) = upstream_channel { + if let Some(hashrate_to_substract) = hashrate_to_substract { + if upstream_channel + .upstream_difficulty + .channel_nominal_hashrate + >= hashrate_to_substract + { + upstream_channel + .upstream_difficulty + .channel_nominal_hashrate -= hashrate_to_substract; + } else { + upstream_channel + .upstream_difficulty + .channel_nominal_hashrate = 0.0; + } + } } - }) - .map_err(|_e| Error::PoisonLock) - })??; + }); + })?; Ok(()) } @@ -98,20 +141,61 @@ impl Downstream { pub async fn try_update_difficulty_settings( self_: Arc>, ) -> ProxyResult<'static, ()> { - let (diff_mgmt, channel_id) = self_ - .clone() - .safe_lock(|d| (d.difficulty_mgmt.clone(), d.connection_id))?; - tracing::debug!( - "Time of last diff update: {:?}", - diff_mgmt.timestamp_of_last_update - ); + let ( + min_individual_miner_hashrate, + timestamp_of_last_update, + submits_since_last_update, + shares_per_minute, + channel_id, + connection_id, + ) = self_.clone().safe_lock(|d| { + let ( + min_individual_miner_hashrate, + timestamp_of_last_update, + submits_since_last_update, + shares_per_minute, + ) = d + .upstream_channel_manager + .super_safe_lock(|upstream_channel_manager| { + let hasrate = upstream_channel_manager + .downstream_difficulty_hashrate(d.channel_id, d.connection_id) + .unwrap(); + let timestamp = upstream_channel_manager + .downstream_difficulty_timestamp_of_last_update( + d.channel_id, + d.connection_id, + ) + .unwrap(); + let submit_since_last_update = upstream_channel_manager + .downstream_difficulty_submits_since_last_update( + d.channel_id, + d.connection_id, + ) + .unwrap(); + ( + hasrate, + timestamp, + submit_since_last_update, + upstream_channel_manager.shares_per_minute, + ) + }); + ( + min_individual_miner_hashrate, + timestamp_of_last_update, + submits_since_last_update, + shares_per_minute, + d.channel_id, + d.connection_id, + ) + })?; + tracing::debug!("Time of last diff update: {:?}", timestamp_of_last_update); tracing::debug!( "Number of shares submitted: {:?}", - diff_mgmt.submits_since_last_update + submits_since_last_update ); let prev_target = match roles_logic_sv2::utils::hash_rate_to_target( - diff_mgmt.min_individual_miner_hashrate.into(), - diff_mgmt.shares_per_minute.into(), + min_individual_miner_hashrate.into(), + shares_per_minute.into(), ) { Ok(target) => target.to_vec(), Err(v) => return Err(Error::TargetError(v)), @@ -121,7 +205,7 @@ impl Downstream { { let new_target = match roles_logic_sv2::utils::hash_rate_to_target( new_hash_rate.into(), - diff_mgmt.shares_per_minute.into(), + shares_per_minute.into(), ) { Ok(target) => target, Err(v) => return Err(Error::TargetError(v)), @@ -131,6 +215,7 @@ impl Downstream { // send mining.set_difficulty to miner Downstream::send_message_downstream(self_.clone(), message).await?; let update_target_msg = SetDownstreamTarget { + connection_id, channel_id, new_target: new_target.into(), }; @@ -153,9 +238,17 @@ impl Downstream { #[allow(clippy::result_large_err)] pub fn hash_rate_to_target(self_: Arc>) -> ProxyResult<'static, Vec> { self_.safe_lock(|d| { + let (min_individual_miner_hashrate, shares_per_minute) = + d.upstream_channel_manager.super_safe_lock(|u| { + let hashrate = u + .downstream_difficulty_hashrate(d.channel_id, d.connection_id) + .unwrap(); + let shares_per_minute = u.shares_per_minute; + (hashrate, shares_per_minute) + }); match roles_logic_sv2::utils::hash_rate_to_target( - d.difficulty_mgmt.min_individual_miner_hashrate.into(), - d.difficulty_mgmt.shares_per_minute.into(), + min_individual_miner_hashrate.into(), + shares_per_minute.into(), ) { Ok(target) => Ok(target.to_vec()), Err(e) => Err(Error::TargetError(e)), @@ -171,7 +264,17 @@ impl Downstream { #[allow(clippy::result_large_err)] pub(super) fn save_share(self_: Arc>) -> ProxyResult<'static, ()> { self_.safe_lock(|d| { - d.difficulty_mgmt.submits_since_last_update += 1; + _ = d.upstream_channel_manager.safe_lock(|u| { + let submits = u + .downstream_difficulty_submits_since_last_update(d.channel_id, d.connection_id); + if let Some(submits_since_last_update) = submits { + u.set_downstream_difficulty_submits_since_last_update( + d.channel_id, + d.connection_id, + submits_since_last_update + 1, + ); + } + }); })?; Ok(()) } @@ -237,14 +340,23 @@ impl Downstream { .expect("time went backwards") .as_secs(); + let (min_individual_miner_hashrate, timestamp_of_last_update, submits_since_last_update, shares_per_minute ) = d.upstream_channel_manager.super_safe_lock(|upstream_channel_manager| { + let hasrate = upstream_channel_manager.downstream_difficulty_hashrate(d.channel_id, d.connection_id).unwrap(); + let timestamp = upstream_channel_manager.downstream_difficulty_timestamp_of_last_update(d.channel_id, d.connection_id).unwrap(); + let submit_since_last_update = upstream_channel_manager.downstream_difficulty_submits_since_last_update(d.channel_id, d.connection_id).unwrap(); + (hasrate, timestamp, submit_since_last_update, upstream_channel_manager.shares_per_minute) + }); + // reset if timestamp is at 0 - if d.difficulty_mgmt.timestamp_of_last_update == 0 { - d.difficulty_mgmt.timestamp_of_last_update = timestamp_secs; - d.difficulty_mgmt.submits_since_last_update = 0; + if timestamp_of_last_update == 0 { + d.upstream_channel_manager.safe_lock(|u| { + u.set_downstream_difficulty_timestamp_of_last_update(d.channel_id, d.connection_id, timestamp_secs); + u.set_downstream_difficulty_submits_since_last_update(d.channel_id, d.connection_id, 0); + })?; return Ok(None); } - let delta_time = timestamp_secs - d.difficulty_mgmt.timestamp_of_last_update; + let delta_time = timestamp_secs - timestamp_of_last_update; #[cfg(test)] if delta_time == 0 { return Ok(None); @@ -255,7 +367,7 @@ impl Downstream { } tracing::debug!("DELTA TIME: {:?}", delta_time); let realized_share_per_min = - d.difficulty_mgmt.submits_since_last_update as f64 / (delta_time as f64 / 60.0); + submits_since_last_update as f64 / (delta_time as f64 / 60.0); tracing::debug!("REALIZED SHARES PER MINUTE: {:?}", realized_share_per_min); tracing::debug!("CURRENT MINER TARGET: {:?}", miner_target); let mut new_miner_hashrate = match roles_logic_sv2::utils::hash_rate_from_target( @@ -265,14 +377,14 @@ impl Downstream { Ok(hashrate) => hashrate as f32, Err(e) => { tracing::debug!("{:?} -> Probably min_individual_miner_hashrate parameter was not set properly in config file. New hashrate will be automatically adjusted to match the real one.", e); - d.difficulty_mgmt.min_individual_miner_hashrate * realized_share_per_min as f32 / d.difficulty_mgmt.shares_per_minute + min_individual_miner_hashrate * realized_share_per_min as f32 / shares_per_minute } }; let mut hashrate_delta = - new_miner_hashrate - d.difficulty_mgmt.min_individual_miner_hashrate; + new_miner_hashrate - min_individual_miner_hashrate; let hashrate_delta_percentage = (hashrate_delta.abs() - / d.difficulty_mgmt.min_individual_miner_hashrate) + / min_individual_miner_hashrate) * 100.0; tracing::debug!("\nMINER HASHRATE: {:?}", new_miner_hashrate); @@ -283,34 +395,40 @@ impl Downstream { || (hashrate_delta_percentage >= 30.0) && (delta_time >= 240) || (hashrate_delta_percentage >= 15.0) && (delta_time >= 300) { - // realized_share_per_min is 0.0 when d.difficulty_mgmt.submits_since_last_update is 0 + // realized_share_per_min is 0.0 when submits_since_last_update is 0 // so it's safe to compare realized_share_per_min with == 0.0 if realized_share_per_min == 0.0 { new_miner_hashrate = match delta_time { - dt if dt <= 30 => d.difficulty_mgmt.min_individual_miner_hashrate / 1.5, - dt if dt < 60 => d.difficulty_mgmt.min_individual_miner_hashrate / 2.0, - _ => d.difficulty_mgmt.min_individual_miner_hashrate / 3.0, + dt if dt <= 30 => min_individual_miner_hashrate / 1.5, + dt if dt < 60 => min_individual_miner_hashrate / 2.0, + _ => min_individual_miner_hashrate / 3.0, }; hashrate_delta = - new_miner_hashrate - d.difficulty_mgmt.min_individual_miner_hashrate; + new_miner_hashrate - min_individual_miner_hashrate; } if (realized_share_per_min > 0.0) && (hashrate_delta_percentage > 1000.0) { new_miner_hashrate = match delta_time { - dt if dt <= 30 => d.difficulty_mgmt.min_individual_miner_hashrate * 10.0, - dt if dt < 60 => d.difficulty_mgmt.min_individual_miner_hashrate * 5.0, - _ => d.difficulty_mgmt.min_individual_miner_hashrate * 3.0, + dt if dt <= 30 => min_individual_miner_hashrate * 10.0, + dt if dt < 60 => min_individual_miner_hashrate * 5.0, + _ => min_individual_miner_hashrate * 3.0, }; hashrate_delta = - new_miner_hashrate - d.difficulty_mgmt.min_individual_miner_hashrate; + new_miner_hashrate - min_individual_miner_hashrate; } - d.difficulty_mgmt.min_individual_miner_hashrate = new_miner_hashrate; - d.difficulty_mgmt.timestamp_of_last_update = timestamp_secs; - d.difficulty_mgmt.submits_since_last_update = 0; - d.upstream_difficulty_config.super_safe_lock(|c| { - if c.channel_nominal_hashrate + hashrate_delta > 0.0 { - c.channel_nominal_hashrate += hashrate_delta; - } else { - c.channel_nominal_hashrate = 0.0; + d.upstream_channel_manager.safe_lock(|u| { + u.set_downstream_difficulty_hashrate(d.channel_id, d.connection_id, new_miner_hashrate); + u.set_downstream_difficulty_timestamp_of_last_update(d.channel_id, d.connection_id, timestamp_secs); + u.set_downstream_difficulty_submits_since_last_update(d.channel_id, d.connection_id, 0); + })?; + + _ = d.upstream_channel_manager.safe_lock(|upstream_channel_manager| { + let upstream_channel = upstream_channel_manager.upstream_manager.get_mut(&d.channel_id); + if let Some(upstream_channel) = upstream_channel { + if upstream_channel.upstream_difficulty.channel_nominal_hashrate + hashrate_delta> 0.0 { + upstream_channel.upstream_difficulty.channel_nominal_hashrate += hashrate_delta; + } else { + upstream_channel.upstream_difficulty.channel_nominal_hashrate = 0.0; + } } }); Ok(Some(new_miner_hashrate)) @@ -331,203 +449,3 @@ impl Downstream { && aligned.iter().all(|&x| x == 0) } } - -#[cfg(test)] -mod test { - - use crate::config::{DownstreamDifficultyConfig, UpstreamDifficultyConfig}; - use async_channel::unbounded; - use binary_sv2::U256; - use rand::{thread_rng, Rng}; - use roles_logic_sv2::{mining_sv2::Target, utils::Mutex}; - use sha2::{Digest, Sha256}; - use std::{ - sync::Arc, - time::{Duration, Instant}, - }; - - use crate::downstream_sv1::Downstream; - - #[ignore] // as described in issue #988 - #[test] - fn test_diff_management() { - let expected_shares_per_minute = 1000.0; - let total_run_time = std::time::Duration::from_secs(60); - let initial_nominal_hashrate = measure_hashrate(5); - let target = match roles_logic_sv2::utils::hash_rate_to_target( - initial_nominal_hashrate, - expected_shares_per_minute, - ) { - Ok(target) => target, - Err(_) => panic!(), - }; - - let mut share = generate_random_80_byte_array(); - let timer = std::time::Instant::now(); - let mut elapsed = std::time::Duration::from_secs(0); - let mut count = 0; - while elapsed <= total_run_time { - // start hashing util a target is met and submit to - mock_mine(target.clone().into(), &mut share); - elapsed = timer.elapsed(); - count += 1; - } - - let calculated_share_per_min = count as f32 / (elapsed.as_secs_f32() / 60.0); - // This is the error margin for a confidence of 99.99...% given the expect number of shares - // per minute TODO the review the math under it - let error_margin = get_error(expected_shares_per_minute); - let error = (calculated_share_per_min - expected_shares_per_minute as f32).abs(); - assert!( - error <= error_margin as f32, - "Calculated shares per minute are outside the 99.99...% confidence interval. Error: {:?}, Error margin: {:?}, {:?}", error, error_margin,calculated_share_per_min - ); - } - - fn get_error(lambda: f64) -> f64 { - let z_score_99 = 6.0; - z_score_99 * lambda.sqrt() - } - - fn mock_mine(target: Target, share: &mut [u8; 80]) { - let mut hashed: Target = [255_u8; 32].into(); - while hashed > target { - hashed = hash(share); - } - } - - // returns hashrate based on how fast the device hashes over the given duration - fn measure_hashrate(duration_secs: u64) -> f64 { - let mut share = generate_random_80_byte_array(); - let start_time = Instant::now(); - let mut hashes: u64 = 0; - let duration = Duration::from_secs(duration_secs); - - while start_time.elapsed() < duration { - for _ in 0..10000 { - hash(&mut share); - hashes += 1; - } - } - - let elapsed_secs = start_time.elapsed().as_secs_f64(); - - hashes as f64 / elapsed_secs - } - - fn hash(share: &mut [u8; 80]) -> Target { - let nonce: [u8; 8] = share[0..8].try_into().unwrap(); - let mut nonce = u64::from_le_bytes(nonce); - nonce += 1; - share[0..8].copy_from_slice(&nonce.to_le_bytes()); - let hash = Sha256::digest(&share).to_vec(); - let hash: U256<'static> = hash.try_into().unwrap(); - hash.into() - } - - fn generate_random_80_byte_array() -> [u8; 80] { - let mut rng = thread_rng(); - let mut arr = [0u8; 80]; - rng.fill(&mut arr[..]); - arr - } - - #[tokio::test] - async fn test_converge_to_spm_from_low() { - test_converge_to_spm(1.0).await - } - //TODO - //#[tokio::test] - //async fn test_converge_to_spm_from_high() { - // test_converge_to_spm(1_000_000_000_000).await - //} - - async fn test_converge_to_spm(start_hashrate: f64) { - let downstream_conf = DownstreamDifficultyConfig { - min_individual_miner_hashrate: 0.0, // updated below - shares_per_minute: 1000.0, // 1000 shares per minute - submits_since_last_update: 0, - timestamp_of_last_update: 0, // updated below - }; - let upstream_config = UpstreamDifficultyConfig { - channel_diff_update_interval: 60, - channel_nominal_hashrate: 0.0, - timestamp_of_last_update: 0, - should_aggregate: false, - }; - let (tx_sv1_submit, _rx_sv1_submit) = unbounded(); - let (tx_outgoing, _rx_outgoing) = unbounded(); - let mut downstream = Downstream::new( - 1, - vec![], - vec![], - None, - None, - tx_sv1_submit, - tx_outgoing, - false, - 0, - downstream_conf.clone(), - Arc::new(Mutex::new(upstream_config)), - "0".to_string(), - ); - downstream.difficulty_mgmt.min_individual_miner_hashrate = start_hashrate as f32; - - let total_run_time = std::time::Duration::from_secs(10); - let config_shares_per_minute = downstream_conf.shares_per_minute; - let timer = std::time::Instant::now(); - let mut elapsed = std::time::Duration::from_secs(0); - - let expected_nominal_hashrate = measure_hashrate(5); - let expected_target = match roles_logic_sv2::utils::hash_rate_to_target( - expected_nominal_hashrate, - config_shares_per_minute.into(), - ) { - Ok(target) => target, - Err(_) => panic!(), - }; - - let initial_nominal_hashrate = start_hashrate; - let mut initial_target = match roles_logic_sv2::utils::hash_rate_to_target( - initial_nominal_hashrate, - config_shares_per_minute.into(), - ) { - Ok(target) => target, - Err(_) => panic!(), - }; - let downstream = Arc::new(Mutex::new(downstream)); - Downstream::init_difficulty_management(downstream.clone(), initial_target.inner_as_ref()) - .await - .unwrap(); - let mut share = generate_random_80_byte_array(); - while elapsed <= total_run_time { - mock_mine(initial_target.clone().into(), &mut share); - Downstream::save_share(downstream.clone()).unwrap(); - Downstream::try_update_difficulty_settings(downstream.clone()) - .await - .unwrap(); - initial_target = downstream - .safe_lock(|d| { - match roles_logic_sv2::utils::hash_rate_to_target( - d.difficulty_mgmt.min_individual_miner_hashrate.into(), - config_shares_per_minute.into(), - ) { - Ok(target) => target, - Err(_) => panic!(), - } - }) - .unwrap(); - elapsed = timer.elapsed(); - } - let expected_0s = trailing_0s(expected_target.inner_as_ref().to_vec()); - let actual_0s = trailing_0s(initial_target.inner_as_ref().to_vec()); - assert!(expected_0s.abs_diff(actual_0s) <= 1); - } - fn trailing_0s(mut v: Vec) -> usize { - let mut ret = 0; - while v.pop() == Some(0) { - ret += 1; - } - ret - } -} diff --git a/roles/translator/src/lib/downstream_sv1/downstream.rs b/roles/translator/src/lib/downstream_sv1/downstream.rs index 3960039baa..d70f5eacf5 100644 --- a/roles/translator/src/lib/downstream_sv1/downstream.rs +++ b/roles/translator/src/lib/downstream_sv1/downstream.rs @@ -19,8 +19,7 @@ //! ([`IsMiningDownstream`], [`IsDownstream`]). use crate::{ - config::{DownstreamDifficultyConfig, UpstreamDifficultyConfig}, - downstream_sv1, + channel_manager::{Sv1ChannelId, UpstreamChannelManager}, error::ProxyResult, status, }; @@ -34,12 +33,9 @@ use tokio::{ task::AbortHandle, }; -use super::{kill, DownstreamMessages, SubmitShareWithChannelId, SUBSCRIBE_TIMEOUT_SECS}; +use super::{kill, DownstreamMessages, SUBSCRIBE_TIMEOUT_SECS}; -use roles_logic_sv2::{ - common_properties::{IsDownstream, IsMiningDownstream}, - utils::Mutex, -}; +use roles_logic_sv2::utils::Mutex; use crate::error::Error; use futures::select; @@ -47,12 +43,7 @@ use tokio_util::codec::{FramedRead, LinesCodec}; use std::{net::SocketAddr, sync::Arc}; use tracing::{debug, info, warn}; -use v1::{ - client_to_server::{self, Submit}, - json_rpc, server_to_client, - utils::{Extranonce, HexU32Be}, - IsServer, -}; +use v1::{client_to_server::Submit, json_rpc, server_to_client, utils::HexU32Be, IsServer}; /// The maximum allowed length for a single line (JSON-RPC message) received from an SV1 client. const MAX_LINE_LENGTH: usize = 2_usize.pow(16); @@ -62,65 +53,34 @@ const MAX_LINE_LENGTH: usize = 2_usize.pow(16); #[derive(Debug)] pub struct Downstream { /// The unique identifier assigned to this downstream connection/channel. - pub(super) connection_id: u32, + pub(super) connection_id: Sv1ChannelId, + /// The channel id of the upstream channel + pub(super) channel_id: u32, /// List of authorized Downstream Mining Devices. - authorized_names: Vec, + pub(super) authorized_names: Vec, /// The extranonce1 value assigned to this downstream miner. - extranonce1: Vec, + pub(super) extranonce1: Vec, /// `extranonce1` to be sent to the Downstream in the SV1 `mining.subscribe` message response. //extranonce1: Vec, //extranonce2_size: usize, /// Version rolling mask bits - version_rolling_mask: Option, + pub(super) version_rolling_mask: Option, /// Minimum version rolling mask bits size - version_rolling_min_bit: Option, + pub(super) version_rolling_min_bit: Option, /// Sends a SV1 `mining.submit` message received from the Downstream role to the `Bridge` for /// translation into a SV2 `SubmitSharesExtended`. - tx_sv1_bridge: Sender, + pub(super) tx_sv1_bridge: Sender, /// Sends message to the SV1 Downstream role. tx_outgoing: Sender, /// True if this is the first job received from `Upstream`. - first_job_received: bool, + pub(super) first_job_received: bool, /// The expected size of the extranonce2 field provided by the miner. - extranonce2_len: usize, - /// Configuration and state for managing difficulty adjustments specific - /// to this individual downstream miner. - pub(super) difficulty_mgmt: DownstreamDifficultyConfig, - /// Configuration settings for the upstream channel's difficulty management. - pub(super) upstream_difficulty_config: Arc>, + pub(super) extranonce2_len: usize, + + pub(super) upstream_channel_manager: Arc>, } impl Downstream { - // not huge fan of test specific code in codebase. - #[cfg(test)] - pub fn new( - connection_id: u32, - authorized_names: Vec, - extranonce1: Vec, - version_rolling_mask: Option, - version_rolling_min_bit: Option, - tx_sv1_bridge: Sender, - tx_outgoing: Sender, - first_job_received: bool, - extranonce2_len: usize, - difficulty_mgmt: DownstreamDifficultyConfig, - upstream_difficulty_config: Arc>, - last_job_id: String, - ) -> Self { - Downstream { - connection_id, - authorized_names, - extranonce1, - version_rolling_mask, - version_rolling_min_bit, - tx_sv1_bridge, - tx_outgoing, - first_job_received, - extranonce2_len, - difficulty_mgmt, - upstream_difficulty_config, - } - } /// Instantiates and manages a new handler for a single downstream SV1 client connection. /// /// This is the primary function called for each new incoming TCP stream from a miner. @@ -134,7 +94,8 @@ impl Downstream { #[allow(clippy::too_many_arguments)] pub async fn new_downstream( stream: TcpStream, - connection_id: u32, + channel_id: u32, + connection_id: Sv1ChannelId, tx_sv1_bridge: Sender, mut rx_sv1_notify: broadcast::Receiver>, tx_status: status::Sender, @@ -142,8 +103,7 @@ impl Downstream { last_notify: Option>, extranonce2_len: usize, host: String, - difficulty_config: DownstreamDifficultyConfig, - upstream_difficulty_config: Arc>, + upstream_channel_manager: Arc>, task_collector: Arc>>, ) { // Reads and writes from Downstream SV1 Mining Device Client @@ -152,6 +112,7 @@ impl Downstream { let downstream = Arc::new(Mutex::new(Downstream { connection_id, + channel_id, authorized_names: vec![], extranonce1, //extranonce1: extranonce1.to_vec(), @@ -161,8 +122,7 @@ impl Downstream { tx_outgoing, first_job_received: false, extranonce2_len, - difficulty_mgmt: difficulty_config, - upstream_difficulty_config, + upstream_channel_manager, })); let self_ = downstream.clone(); @@ -389,9 +349,8 @@ impl Downstream { tx_mining_notify: broadcast::Sender>, tx_status: status::Sender, bridge: Arc>, - downstream_difficulty_config: DownstreamDifficultyConfig, - upstream_difficulty_config: Arc>, task_collector: Arc>>, + upstream_channel_manager: Arc>, ) { let accept_connections = tokio::task::spawn({ let task_collector = task_collector.clone(); @@ -399,20 +358,17 @@ impl Downstream { let listener = TcpListener::bind(downstream_addr).await.unwrap(); while let Ok((stream, _)) = listener.accept().await { - let expected_hash_rate = - downstream_difficulty_config.min_individual_miner_hashrate; - let open_sv1_downstream = bridge - .safe_lock(|s| s.on_new_sv1_connection(expected_hash_rate)) - .unwrap(); - + let mut bridge = bridge.safe_lock(|s| s.clone()).unwrap(); + let open_sv1_downstream = bridge.on_new_sv1_connection().await; let host = stream.peer_addr().unwrap().to_string(); match open_sv1_downstream { - Ok(opened) => { + Some(opened) => { info!("PROXY SERVER - ACCEPTING FROM DOWNSTREAM: {}", host); Downstream::new_downstream( stream, opened.channel_id, + opened.connection_id, tx_sv1_submit.clone(), tx_mining_notify.subscribe(), tx_status.listener_to_connection(), @@ -420,17 +376,13 @@ impl Downstream { opened.last_notify, opened.extranonce2_len as usize, host, - downstream_difficulty_config.clone(), - upstream_difficulty_config.clone(), + upstream_channel_manager.clone(), task_collector.clone(), ) .await; } - Err(e) => { - tracing::error!( - "Failed to create a new downstream connection: {:?}", - e - ); + None => { + tracing::error!("Failed to create a new downstream connection",); } } } @@ -512,199 +464,3 @@ impl Downstream { Ok(()) } } - -/// Implements `IsServer` for `Downstream` to handle the SV1 messages. -impl IsServer<'static> for Downstream { - /// Handles the incoming SV1 `mining.configure` message. - /// - /// This message is received after `mining.subscribe` and `mining.authorize`. - /// It allows the miner to negotiate capabilities, particularly regarding - /// version rolling. This method processes the version rolling mask and - /// minimum bit count provided by the client. - /// - /// Returns a tuple containing: - /// 1. `Option`: The version rolling parameters - /// negotiated by the server (proxy). - /// 2. `Option`: A boolean indicating whether the server (proxy) supports version rolling - /// (always `Some(false)` for TProxy according to the SV1 spec when not supporting work - /// selection). - fn handle_configure( - &mut self, - request: &client_to_server::Configure, - ) -> (Option, Option) { - info!("Down: Configuring"); - debug!("Down: Handling mining.configure: {:?}", &request); - - // TODO 0x1FFFE000 should be configured - // = 11111111111111110000000000000 - // this is a reasonable default as it allows all 16 version bits to be used - // If the tproxy/pool needs to use some version bits this needs to be configurable - // so upstreams can negotiate with downstreams. When that happens this should consider - // the min_bit_count in the mining.configure message - self.version_rolling_mask = request - .version_rolling_mask() - .map(|mask| HexU32Be(mask & 0x1FFFE000)); - self.version_rolling_min_bit = request.version_rolling_min_bit_count(); - - debug!( - "Negotiated version_rolling_mask is {:?}", - self.version_rolling_mask - ); - ( - Some(server_to_client::VersionRollingParams::new( - self.version_rolling_mask.clone().unwrap_or(HexU32Be(0)), - self.version_rolling_min_bit.clone().unwrap_or(HexU32Be(0)), - ).expect("Version mask invalid, automatic version mask selection not supported, please change it in carte::downstream_sv1::mod.rs")), - Some(false), - ) - } - - /// Handles the incoming SV1 `mining.subscribe` message. - /// - /// This is typically the first message received from a new client. In the SV1 - /// protocol, it's used to subscribe to job notifications and receive session - /// details like extranonce1 and extranonce2 size. This method acknowledges the subscription and - /// provides the necessary details derived from the upstream SV2 connection (extranonce1 and - /// extranonce2 size). It also provides subscription IDs for the - /// `mining.set_difficulty` and `mining.notify` methods. - fn handle_subscribe(&self, request: &client_to_server::Subscribe) -> Vec<(String, String)> { - info!("Down: Subscribing"); - debug!("Down: Handling mining.subscribe: {:?}", &request); - - let set_difficulty_sub = ( - "mining.set_difficulty".to_string(), - downstream_sv1::new_subscription_id(), - ); - let notify_sub = ( - "mining.notify".to_string(), - "ae6812eb4cd7735a302a8a9dd95cf71f".to_string(), - ); - - vec![set_difficulty_sub, notify_sub] - } - - /// Any numbers of workers may be authorized at any time during the session. In this way, a - /// large number of independent Mining Devices can be handled with a single SV1 connection. - /// https://bitcoin.stackexchange.com/questions/29416/how-do-pool-servers-handle-multiple-workers-sharing-one-connection-with-stratum - fn handle_authorize(&self, request: &client_to_server::Authorize) -> bool { - info!("Down: Authorizing"); - debug!("Down: Handling mining.authorize: {:?}", &request); - true - } - - /// Handles the incoming SV1 `mining.submit` message. - /// - /// This message is sent by the miner when they find a share that meets - /// their current difficulty target. It contains the job ID, ntime, nonce, - /// and extranonce2. - /// - /// This method processes the submitted share, potentially validates it - /// against the downstream target (although this might happen in the Bridge - /// or difficulty management logic), translates it into a - /// [`SubmitShareWithChannelId`], and sends it to the Bridge for - /// translation to SV2 and forwarding upstream if it meets the upstream target. - fn handle_submit(&self, request: &client_to_server::Submit<'static>) -> bool { - info!("Down: Submitting Share {:?}", request); - debug!("Down: Handling mining.submit: {:?}", &request); - - // TODO: Check if receiving valid shares by adding diff field to Downstream - - let to_send = SubmitShareWithChannelId { - channel_id: self.connection_id, - share: request.clone(), - extranonce: self.extranonce1.clone(), - extranonce2_len: self.extranonce2_len, - version_rolling_mask: self.version_rolling_mask.clone(), - }; - - self.tx_sv1_bridge - .try_send(DownstreamMessages::SubmitShares(to_send)) - .unwrap(); - - true - } - - /// Indicates to the server that the client supports the mining.set_extranonce method. - fn handle_extranonce_subscribe(&self) {} - - /// Checks if a Downstream role is authorized. - fn is_authorized(&self, name: &str) -> bool { - self.authorized_names.contains(&name.to_string()) - } - - /// Authorizes a Downstream role. - fn authorize(&mut self, name: &str) { - self.authorized_names.push(name.to_string()); - } - - /// Sets the `extranonce1` field sent in the SV1 `mining.notify` message to the value specified - /// by the SV2 `OpenExtendedMiningChannelSuccess` message sent from the Upstream role. - fn set_extranonce1( - &mut self, - _extranonce1: Option>, - ) -> Extranonce<'static> { - self.extranonce1.clone().try_into().unwrap() - } - - /// Returns the `Downstream`'s `extranonce1` value. - fn extranonce1(&self) -> Extranonce<'static> { - self.extranonce1.clone().try_into().unwrap() - } - - /// Sets the `extranonce2_size` field sent in the SV1 `mining.notify` message to the value - /// specified by the SV2 `OpenExtendedMiningChannelSuccess` message sent from the Upstream role. - fn set_extranonce2_size(&mut self, _extra_nonce2_size: Option) -> usize { - self.extranonce2_len - } - - /// Returns the `Downstream`'s `extranonce2_size` value. - fn extranonce2_size(&self) -> usize { - self.extranonce2_len - } - - /// Returns the version rolling mask. - fn version_rolling_mask(&self) -> Option { - self.version_rolling_mask.clone() - } - - /// Sets the version rolling mask. - fn set_version_rolling_mask(&mut self, mask: Option) { - self.version_rolling_mask = mask; - } - - /// Sets the minimum version rolling bit. - fn set_version_rolling_min_bit(&mut self, mask: Option) { - self.version_rolling_min_bit = mask - } - - fn notify(&mut self) -> Result { - unreachable!() - } -} - -// Can we remove this? -impl IsMiningDownstream for Downstream {} -// Can we remove this? -impl IsDownstream for Downstream { - fn get_downstream_mining_data( - &self, - ) -> roles_logic_sv2::common_properties::CommonDownstreamData { - todo!() - } -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn gets_difficulty_from_target() { - let target = vec![ - 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 128, 255, 127, - 0, 0, 0, 0, 0, - ]; - let actual = Downstream::difficulty_from_target(target).unwrap(); - let expect = 512.0; - assert_eq!(actual, expect); - } -} diff --git a/roles/translator/src/lib/downstream_sv1/message_handler.rs b/roles/translator/src/lib/downstream_sv1/message_handler.rs new file mode 100644 index 0000000000..a8b6e8f66d --- /dev/null +++ b/roles/translator/src/lib/downstream_sv1/message_handler.rs @@ -0,0 +1,195 @@ +use crate::downstream_sv1; + +use super::{Downstream, DownstreamMessages, SubmitShareWithChannelId}; + +use roles_logic_sv2::common_properties::{IsDownstream, IsMiningDownstream}; + +use tracing::{debug, info}; +use v1::{ + client_to_server, json_rpc, server_to_client, + utils::{Extranonce, HexU32Be}, + IsServer, +}; + +/// Implements `IsServer` for `Downstream` to handle the SV1 messages. +impl IsServer<'static> for Downstream { + /// Handles the incoming SV1 `mining.configure` message. + /// + /// This message is received after `mining.subscribe` and `mining.authorize`. + /// It allows the miner to negotiate capabilities, particularly regarding + /// version rolling. This method processes the version rolling mask and + /// minimum bit count provided by the client. + /// + /// Returns a tuple containing: + /// 1. `Option`: The version rolling parameters + /// negotiated by the server (proxy). + /// 2. `Option`: A boolean indicating whether the server (proxy) supports version rolling + /// (always `Some(false)` for TProxy according to the SV1 spec when not supporting work + /// selection). + fn handle_configure( + &mut self, + request: &client_to_server::Configure, + ) -> (Option, Option) { + info!("Down: Configuring"); + debug!("Down: Handling mining.configure: {:?}", &request); + + // TODO 0x1FFFE000 should be configured + // = 11111111111111110000000000000 + // this is a reasonable default as it allows all 16 version bits to be used + // If the tproxy/pool needs to use some version bits this needs to be configurable + // so upstreams can negotiate with downstreams. When that happens this should consider + // the min_bit_count in the mining.configure message + self.version_rolling_mask = request + .version_rolling_mask() + .map(|mask| HexU32Be(mask & 0x1FFFE000)); + self.version_rolling_min_bit = request.version_rolling_min_bit_count(); + + debug!( + "Negotiated version_rolling_mask is {:?}", + self.version_rolling_mask + ); + ( + Some(server_to_client::VersionRollingParams::new( + self.version_rolling_mask.clone().unwrap_or(HexU32Be(0)), + self.version_rolling_min_bit.clone().unwrap_or(HexU32Be(0)), + ).expect("Version mask invalid, automatic version mask selection not supported, please change it in carte::downstream_sv1::mod.rs")), + Some(false), + ) + } + + /// Handles the incoming SV1 `mining.subscribe` message. + /// + /// This is typically the first message received from a new client. In the SV1 + /// protocol, it's used to subscribe to job notifications and receive session + /// details like extranonce1 and extranonce2 size. This method acknowledges the subscription and + /// provides the necessary details derived from the upstream SV2 connection (extranonce1 and + /// extranonce2 size). It also provides subscription IDs for the + /// `mining.set_difficulty` and `mining.notify` methods. + fn handle_subscribe(&self, request: &client_to_server::Subscribe) -> Vec<(String, String)> { + info!("Down: Subscribing"); + debug!("Down: Handling mining.subscribe: {:?}", &request); + + let set_difficulty_sub = ( + "mining.set_difficulty".to_string(), + downstream_sv1::new_subscription_id(), + ); + let notify_sub = ( + "mining.notify".to_string(), + "ae6812eb4cd7735a302a8a9dd95cf71f".to_string(), + ); + + vec![set_difficulty_sub, notify_sub] + } + + /// Any numbers of workers may be authorized at any time during the session. In this way, a + /// large number of independent Mining Devices can be handled with a single SV1 connection. + /// https://bitcoin.stackexchange.com/questions/29416/how-do-pool-servers-handle-multiple-workers-sharing-one-connection-with-stratum + fn handle_authorize(&self, request: &client_to_server::Authorize) -> bool { + info!("Down: Authorizing"); + debug!("Down: Handling mining.authorize: {:?}", &request); + true + } + + /// Handles the incoming SV1 `mining.submit` message. + /// + /// This message is sent by the miner when they find a share that meets + /// their current difficulty target. It contains the job ID, ntime, nonce, + /// and extranonce2. + /// + /// This method processes the submitted share, potentially validates it + /// against the downstream target (although this might happen in the Bridge + /// or difficulty management logic), translates it into a + /// [`SubmitShareWithChannelId`], and sends it to the Bridge for + /// translation to SV2 and forwarding upstream if it meets the upstream target. + fn handle_submit(&self, request: &client_to_server::Submit<'static>) -> bool { + info!("Down: Submitting Share {:?}", request); + debug!("Down: Handling mining.submit: {:?}", &request); + + // TODO: Check if receiving valid shares by adding diff field to Downstream + + let (tx, _rx) = async_channel::unbounded::(); + + let to_send = SubmitShareWithChannelId { + connection_id: self.connection_id, + channel_id: self.channel_id, + share: request.clone(), + extranonce: self.extranonce1.clone(), + extranonce2_len: self.extranonce2_len, + version_rolling_mask: self.version_rolling_mask.clone(), + verdict_sender: tx, + }; + + self.tx_sv1_bridge + .try_send(DownstreamMessages::SubmitShares(to_send)) + .unwrap(); + true + } + + /// Indicates to the server that the client supports the mining.set_extranonce method. + fn handle_extranonce_subscribe(&self) {} + + /// Checks if a Downstream role is authorized. + fn is_authorized(&self, name: &str) -> bool { + self.authorized_names.contains(&name.to_string()) + } + + /// Authorizes a Downstream role. + fn authorize(&mut self, name: &str) { + self.authorized_names.push(name.to_string()); + } + + /// Sets the `extranonce1` field sent in the SV1 `mining.notify` message to the value specified + /// by the SV2 `OpenExtendedMiningChannelSuccess` message sent from the Upstream role. + fn set_extranonce1( + &mut self, + _extranonce1: Option>, + ) -> Extranonce<'static> { + self.extranonce1.clone().try_into().unwrap() + } + + /// Returns the `Downstream`'s `extranonce1` value. + fn extranonce1(&self) -> Extranonce<'static> { + self.extranonce1.clone().try_into().unwrap() + } + + /// Sets the `extranonce2_size` field sent in the SV1 `mining.notify` message to the value + /// specified by the SV2 `OpenExtendedMiningChannelSuccess` message sent from the Upstream role. + fn set_extranonce2_size(&mut self, _extra_nonce2_size: Option) -> usize { + self.extranonce2_len + } + + /// Returns the `Downstream`'s `extranonce2_size` value. + fn extranonce2_size(&self) -> usize { + self.extranonce2_len + } + + /// Returns the version rolling mask. + fn version_rolling_mask(&self) -> Option { + self.version_rolling_mask.clone() + } + + /// Sets the version rolling mask. + fn set_version_rolling_mask(&mut self, mask: Option) { + self.version_rolling_mask = mask; + } + + /// Sets the minimum version rolling bit. + fn set_version_rolling_min_bit(&mut self, mask: Option) { + self.version_rolling_min_bit = mask + } + + fn notify(&mut self) -> Result { + unreachable!() + } +} + +// Can we remove this? +impl IsMiningDownstream for Downstream {} +// Can we remove this? +impl IsDownstream for Downstream { + fn get_downstream_mining_data( + &self, + ) -> roles_logic_sv2::common_properties::CommonDownstreamData { + todo!() + } +} diff --git a/roles/translator/src/lib/downstream_sv1/mod.rs b/roles/translator/src/lib/downstream_sv1/mod.rs index f0847acb92..e6ef3fa2a3 100644 --- a/roles/translator/src/lib/downstream_sv1/mod.rs +++ b/roles/translator/src/lib/downstream_sv1/mod.rs @@ -11,12 +11,16 @@ //! - [`diff_management`]: (Declared here, likely contains downstream difficulty logic) //! - [`downstream`]: Defines the core [`Downstream`] struct and its functionalities. +use async_channel::Sender; use roles_logic_sv2::mining_sv2::Target; use v1::{client_to_server::Submit, utils::HexU32Be}; pub mod diff_management; pub mod downstream; +pub mod message_handler; pub use downstream::Downstream; +use crate::channel_manager::Sv1ChannelId; + /// This constant defines a timeout duration. It is used to enforce /// that clients sending a `mining.subscribe` message must follow up /// with a `mining.authorize` within this period. This prevents @@ -37,13 +41,15 @@ pub enum DownstreamMessages { /// wrapper around a `mining.submit` with extra channel informationfor the Bridge to /// process -#[derive(Debug)] +#[derive(Debug, Clone)] pub struct SubmitShareWithChannelId { + pub connection_id: Sv1ChannelId, pub channel_id: u32, pub share: Submit<'static>, pub extranonce: Vec, pub extranonce2_len: usize, pub version_rolling_mask: Option, + pub verdict_sender: Sender, } /// message for notifying the bridge that a downstream target has updated @@ -51,6 +57,7 @@ pub struct SubmitShareWithChannelId { #[derive(Debug)] pub struct SetDownstreamTarget { pub channel_id: u32, + pub connection_id: Sv1ChannelId, pub new_target: Target, } diff --git a/roles/translator/src/lib/mod.rs b/roles/translator/src/lib/mod.rs index 26eca7dc25..34ec99ea85 100644 --- a/roles/translator/src/lib/mod.rs +++ b/roles/translator/src/lib/mod.rs @@ -11,6 +11,7 @@ //! It relies on several sub-modules (`config`, `downstream_sv1`, `upstream_sv2`, `proxy`, `status`, //! etc.) for specialized functionalities. use async_channel::{bounded, unbounded}; +use channel_manager::UpstreamChannelManager; use futures::FutureExt; use rand::Rng; pub use roles_logic_sv2::utils::Mutex; @@ -33,6 +34,7 @@ use config::TranslatorConfig; use crate::status::State; +pub mod channel_manager; pub mod config; pub mod downstream_sv1; pub mod error; @@ -49,6 +51,12 @@ pub struct TranslatorSv2 { shutdown: Arc, } +#[derive(Clone, Debug)] +pub struct OpenConnection { + pub request_id: u32, + pub user_identity: String, +} + impl TranslatorSv2 { /// Creates a new `TranslatorSv2`. /// @@ -72,9 +80,6 @@ impl TranslatorSv2 { // Status channel for components to signal errors or state changes. let (tx_status, rx_status) = unbounded(); - // Shared mutable state for the current mining target. - let target = Arc::new(Mutex::new(vec![0; 32])); - // Broadcast channel to send SV1 `mining.notify` messages from the Bridge // to all connected Downstream (SV1) clients. let (tx_sv1_notify, _rx_sv1_notify): ( @@ -91,7 +96,6 @@ impl TranslatorSv2 { Self::internal_start( self.config.clone(), tx_sv1_notify.clone(), - target.clone(), tx_status.clone(), task_collector.clone(), ) @@ -133,7 +137,6 @@ impl TranslatorSv2 { error!("Trying to reconnect the Upstream because of: {}", err); let task_collector1 = task_collector_.clone(); let tx_sv1_notify1 = tx_sv1_notify.clone(); - let target = target.clone(); let tx_status = tx_status.clone(); let proxy_config = self.config.clone(); // Spawn a new task to handle the reconnection process. @@ -150,7 +153,6 @@ impl TranslatorSv2 { Self::internal_start( proxy_config, tx_sv1_notify1, - target.clone(), tx_status.clone(), task_collector1, ) @@ -189,7 +191,6 @@ impl TranslatorSv2 { async fn internal_start( proxy_config: TranslatorConfig, tx_sv1_notify: broadcast::Sender>, - target: Arc>>, tx_status: async_channel::Sender>, task_collector: Arc>>, ) { @@ -202,12 +203,11 @@ impl TranslatorSv2 { // Channel: Upstream -> Bridge (SV2 NewExtendedMiningJob) let (tx_sv2_new_ext_mining_job, rx_sv2_new_ext_mining_job) = bounded(10); - // Channel: Upstream -> internal_start -> Bridge (Initial Extranonce) - let (tx_sv2_extranonce, rx_sv2_extranonce) = bounded(1); - // Channel: Upstream -> Bridge (SV2 SetNewPrevHash) let (tx_sv2_set_new_prev_hash, rx_sv2_set_new_prev_hash) = bounded(10); + let (tx_open_upstream_channel, rx_open_upstream_channel) = bounded::(10); + // Prepare upstream connection address. let upstream_addr = SocketAddr::new( IpAddr::from_str(&proxy_config.upstream_address) @@ -215,8 +215,17 @@ impl TranslatorSv2 { proxy_config.upstream_port, ); - // Shared difficulty configuration - let diff_config = Arc::new(Mutex::new(proxy_config.upstream_difficulty_config.clone())); + let upstream_channel_manager = Arc::new(Mutex::new(UpstreamChannelManager::new( + proxy_config.min_extranonce2_size, + proxy_config + .upstream_difficulty_config + .channel_nominal_hashrate, + proxy_config + .upstream_difficulty_config + .channel_diff_update_interval, + proxy_config.downstream_difficulty_config.shares_per_minute, + ))); + let task_collector_upstream = task_collector.clone(); // Instantiate the Upstream (SV2) component. let upstream = match upstream_sv2::Upstream::new( @@ -225,12 +234,12 @@ impl TranslatorSv2 { rx_sv2_submit_shares_ext, // Receives shares from Bridge tx_sv2_set_new_prev_hash, // Sends prev hash updates to Bridge tx_sv2_new_ext_mining_job, // Sends new jobs to Bridge - proxy_config.min_extranonce2_size, - tx_sv2_extranonce, // Sends initial extranonce - status::Sender::Upstream(tx_status.clone()), // Sends status updates - target.clone(), // Shares target state - diff_config.clone(), // Shares difficulty config + status::Sender::Upstream(tx_status.clone()), // Shares target state task_collector_upstream, + upstream_channel_manager.clone(), + proxy_config.min_supported_version, + proxy_config.max_supported_version, + rx_open_upstream_channel, ) .await { @@ -248,21 +257,7 @@ impl TranslatorSv2 { // even during potentially long-running connection attempts. let task = task::spawn(async move { // Connect to the SV2 Upstream role - match upstream_sv2::Upstream::connect( - upstream.clone(), - proxy_config.min_supported_version, - proxy_config.max_supported_version, - ) - .await - { - Ok(_) => info!("Connected to Upstream!"), - Err(e) => { - // FIXME: Send error to status main loop, and then exit. - error!("Failed to connect to Upstream EXITING! : {}", e); - return; - } - } - + _ = upstream_sv2::Upstream::connect(upstream.clone()).await; // Start the task to parse incoming messages from the Upstream. if let Err(e) = upstream_sv2::Upstream::parse_incoming(upstream.clone()) { error!("failed to create sv2 parser: {}", e); @@ -276,17 +271,6 @@ impl TranslatorSv2 { return; } - // Wait to receive the initial extranonce information from the Upstream. - // This is needed before the Bridge can be fully initialized. - let (extended_extranonce, up_id) = rx_sv2_extranonce.recv().await.unwrap(); - loop { - let target: [u8; 32] = target.safe_lock(|t| t.clone()).unwrap().try_into().unwrap(); - if target != [0; 32] { - break; - }; - tokio::time::sleep(std::time::Duration::from_millis(100)).await; - } - let task_collector_bridge = task_collector_init_task.clone(); // Instantiate the Bridge component. let b = proxy::Bridge::new( @@ -296,10 +280,10 @@ impl TranslatorSv2 { rx_sv2_new_ext_mining_job, tx_sv1_notify.clone(), status::Sender::Bridge(tx_status.clone()), - extended_extranonce, - target, - up_id, task_collector_bridge, + upstream_channel_manager.clone(), + tx_open_upstream_channel, + true, ); // Start the Bridge's main processing loop. proxy::Bridge::start(b.clone()); @@ -318,9 +302,8 @@ impl TranslatorSv2 { tx_sv1_notify, status::Sender::DownstreamListener(tx_status.clone()), b, - proxy_config.downstream_difficulty_config, - diff_config, task_collector_downstream, + upstream_channel_manager.clone(), ); }); // End of init task let _ = diff --git a/roles/translator/src/lib/proxy/bridge.rs b/roles/translator/src/lib/proxy/bridge.rs index 5790d31d5a..f0859df61e 100644 --- a/roles/translator/src/lib/proxy/bridge.rs +++ b/roles/translator/src/lib/proxy/bridge.rs @@ -17,37 +17,39 @@ //! - Broadcasting translated SV1 notifications to connected downstream miners. //! - Managing channel state and difficulty related to job translation. //! - Handling new downstream SV1 connections. +use crate::{ + channel_manager::{Sv1ChannelId, UpstreamChannelManager}, + proxy::next_mining_notify::create_notify, + OpenConnection, +}; + use super::super::{ downstream_sv1::{DownstreamMessages, SetDownstreamTarget, SubmitShareWithChannelId}, - error::{ - Error::{self, PoisonLock}, - ProxyResult, - }, + error::{Error, ProxyResult}, status, }; use async_channel::{Receiver, Sender}; use error_handling::handle_result; use roles_logic_sv2::{ - channel_logic::channel_factory::{ - ExtendedChannelKind, OnNewShare, ProxyExtendedChannelFactory, Share, - }, - mining_sv2::{ - ExtendedExtranonce, NewExtendedMiningJob, SetNewPrevHash, SubmitSharesExtended, Target, - }, - parsers::Mining, - utils::{GroupId, Mutex}, + mining_sv2::{NewExtendedMiningJob, SetNewPrevHash, SubmitSharesExtended}, + utils::Mutex, Error as RolesLogicError, }; -use std::sync::Arc; -use tokio::{sync::broadcast, task::AbortHandle}; -use tracing::{debug, error, info, warn}; +use std::{ + sync::{atomic::AtomicU32, Arc}, + time::Duration, +}; +use tokio::{sync::broadcast, task::AbortHandle, time::sleep}; +use tracing::{debug, info, warn}; use v1::{client_to_server::Submit, server_to_client, utils::HexU32Be}; +static REQUEST_ID: AtomicU32 = AtomicU32::new(0); + /// Bridge between the SV2 `Upstream` and SV1 `Downstream` responsible for the following messaging /// translation: /// 1. SV1 `mining.submit` -> SV2 `SubmitSharesExtended` /// 2. SV2 `SetNewPrevHash` + `NewExtendedMiningJob` -> SV1 `mining.notify` -#[derive(Debug)] +#[derive(Debug, Clone)] pub struct Bridge { /// Receives a SV1 `mining.submit` message from the Downstream role. rx_sv1_downstream: Receiver, @@ -67,31 +69,10 @@ pub struct Bridge { /// Allows the bridge the ability to communicate back to the main thread any status updates /// that would interest the main thread for error handling tx_status: status::Sender, - /// Stores the most recent SV1 `mining.notify` values to be sent to the `Downstream` upon - /// receiving a new SV2 `SetNewPrevHash` and `NewExtendedMiningJob` messages **before** any - /// Downstream role connects to the proxy. - /// - /// Once the proxy establishes a connection with the SV2 Upstream role, it immediately receives - /// a SV2 `SetNewPrevHash` and `NewExtendedMiningJob` message. This happens before the - /// connection to the Downstream role(s) occur. The `last_notify` member fields allows these - /// first notify values to be relayed to the `Downstream` once a Downstream role connects. Once - /// a Downstream role connects and receives the first notify values, this member field is no - /// longer used. - last_notify: Option>, - pub(self) channel_factory: ProxyExtendedChannelFactory, - /// Stores `NewExtendedMiningJob` messages received from the upstream with the `is_future` flag - /// set. These jobs are buffered until a corresponding `SetNewPrevHash` message is - /// received. - future_jobs: Vec>, - /// Stores the last received SV2 `SetNewPrevHash` message. Used in conjunction with - /// `future_jobs` to construct `mining.notify` messages. - last_p_hash: Option>, - /// The mining target currently in use by the downstream miners connected to this bridge. - /// This target is derived from the upstream's requirements but may be adjusted locally. - target: Arc>>, - /// The job ID of the last sent `mining.notify` message. - last_job_id: u32, task_collector: Arc>>, + upstream_channel_manager: Arc>, + tx_open_upstream_channel: Sender, + aggregate: bool, } impl Bridge { @@ -108,16 +89,11 @@ impl Bridge { rx_sv2_new_ext_mining_job: Receiver>, tx_sv1_notify: broadcast::Sender>, tx_status: status::Sender, - extranonces: ExtendedExtranonce, - target: Arc>>, - up_id: u32, task_collector: Arc>>, + upstream_channel_manager: Arc>, + tx_open_upstream_channel: Sender, + aggregate: bool, ) -> Arc> { - let ids = Arc::new(Mutex::new(GroupId::new())); - let share_per_min = 1.0; - let upstream_target: [u8; 32] = - target.safe_lock(|t| t.clone()).unwrap().try_into().unwrap(); - let upstream_target: Target = upstream_target.into(); Arc::new(Mutex::new(Self { rx_sv1_downstream, tx_sv2_submit_shares_ext, @@ -125,24 +101,81 @@ impl Bridge { rx_sv2_new_ext_mining_job, tx_sv1_notify, tx_status, - last_notify: None, - channel_factory: ProxyExtendedChannelFactory::new( - ids, - extranonces, - None, - share_per_min, - ExtendedChannelKind::Proxy { upstream_target }, - None, - up_id, - ), - future_jobs: vec![], - last_p_hash: None, - target, - last_job_id: 0, task_collector, + upstream_channel_manager, + tx_open_upstream_channel, + aggregate, })) } + pub fn have_upstream_channel(&mut self, request_id: u32) -> Option { + let result = self + .upstream_channel_manager + .safe_lock(|upstream_channel_manager| { + info!( + "Received new downstream sv1 connection, number of upstream channels: {:?}", + upstream_channel_manager.upstream_manager.len() + ); + + let channel_id = upstream_channel_manager + .request_id_to_channel_id + .get(&request_id)?; + + let upstream_channel = upstream_channel_manager + .upstream_manager + .get_mut(channel_id)?; + + let (channel_id, connection_id, extranonce, extranonce2_len) = upstream_channel + .downstream_manager + .on_new_downstream_connection(format!("{:?}:miner", request_id)); + debug!("{channel_id:?}, {connection_id:?}, {extranonce:?}, {extranonce2_len:?}"); + let active_job = upstream_channel.downstream_manager.active_job.clone(); + let prev_hash = upstream_channel.downstream_manager.prev_block_hash.clone(); + if let Some(active_job) = active_job { + let result = prev_hash.map(|m| { + let last_notify = create_notify(m, active_job, true); + OpenSv1Downstream { + channel_id, + connection_id, + last_notify: Some(last_notify), + extranonce, + extranonce2_len: extranonce2_len as u16, + } + }); + return result; + } + Some(OpenSv1Downstream { + channel_id, + connection_id, + last_notify: None, + extranonce, + extranonce2_len: extranonce2_len as u16, + }) + }) + .unwrap(); + result + } + + pub async fn open_channel_upstream(&mut self) -> u32 { + info!("Opening new channel with upstream"); + let request_id = REQUEST_ID.fetch_add(1, std::sync::atomic::Ordering::SeqCst) + 1; + while let Err(send_error) = self + .tx_open_upstream_channel + .send(OpenConnection { + request_id, + user_identity: format!("{:?}:miner", request_id), + }) + .await + { + warn!( + "Received an error while sending the open channel request: {:?}", + send_error + ); + } + info!("Successful sent open connection request to upstream subsystem"); + request_id + } + /// Handles the event of a new SV1 downstream client connecting. /// /// Creates a new extended channel using the internal `channel_factory` for the @@ -150,42 +183,32 @@ impl Bridge { /// extranonce and target for the miner, and provides the last known /// `mining.notify` message to immediately send to the new client. #[allow(clippy::result_large_err)] - pub fn on_new_sv1_connection( - &mut self, - hash_rate: f32, - ) -> ProxyResult<'static, OpenSv1Downstream> { - match self.channel_factory.new_extended_channel(0, hash_rate, 0) { - Ok(messages) => { - for message in messages { - match message { - Mining::OpenExtendedMiningChannelSuccess(success) => { - let extranonce = success.extranonce_prefix.to_vec(); - let extranonce2_len = success.extranonce_size; - self.target.safe_lock(|t| *t = success.target.to_vec())?; - return Ok(OpenSv1Downstream { - channel_id: success.channel_id, - last_notify: self.last_notify.clone(), - extranonce, - target: self.target.clone(), - extranonce2_len, - }); - } - Mining::OpenMiningChannelError(_) => todo!(), - Mining::SetNewPrevHash(_) => (), - Mining::NewExtendedMiningJob(_) => (), - _ => unreachable!(), - } + pub async fn on_new_sv1_connection(&mut self) -> Option { + let aggregate = self.aggregate; + if aggregate { + let current_request_id = REQUEST_ID.load(std::sync::atomic::Ordering::Relaxed); + + if let Some(open_sv1_downstream) = self.have_upstream_channel(current_request_id) { + return Some(open_sv1_downstream); + } + + let request_id = self.open_channel_upstream().await; + + loop { + sleep(Duration::from_secs(1)).await; + if let Some(open_sv1_downstream) = self.have_upstream_channel(request_id) { + return Some(open_sv1_downstream); } } - Err(_) => { - return Err(Error::SubprotocolMining( - "Bridge: failed to open new extended channel".to_string(), - )) + } else { + let request_id = self.open_channel_upstream().await; + loop { + sleep(Duration::from_secs(1)).await; + if let Some(open_sv1_downstream) = self.have_upstream_channel(request_id) { + return Some(open_sv1_downstream); + } } - }; - Err(Error::SubprotocolMining( - "Bridge: Invalid mining message when opening downstream connection".to_string(), - )) + } } /// Starts the tasks responsible for receiving and processing @@ -197,48 +220,93 @@ impl Bridge { /// 3. `handle_downstream_messages`: Listens for `DownstreamMessages` (e.g., submit shares) from /// downstream clients. pub fn start(self_: Arc>) { - Self::handle_new_prev_hash(self_.clone()); - Self::handle_new_extended_mining_job(self_.clone()); + Self::start_upstream_job_handler(self_.clone()); Self::handle_downstream_messages(self_); } + fn start_upstream_job_handler(self_: Arc>) { + let task_collector = self_.safe_lock(|b| b.task_collector.clone()).unwrap(); + let (tx_sv1_notify, rx_prev_hash, rx_new_job, tx_status) = self_ + .safe_lock(|s| { + ( + s.tx_sv1_notify.clone(), + s.rx_sv2_set_new_prev_hash.clone(), + s.rx_sv2_new_ext_mining_job.clone(), + s.tx_status.clone(), + ) + }) + .unwrap(); + + debug!("Starting upstream job handler task"); + let handle = tokio::task::spawn(async move { + loop { + tokio::select! { + Ok(prev_hash) = rx_prev_hash.recv() => { + debug!("Received SetNewPrevHash (Job ID: {:?})", prev_hash.job_id); + handle_result!( + tx_status.clone(), + Self::handle_new_prev_hash_(self_.clone(), prev_hash, tx_sv1_notify.clone()).await + ); + } + Ok(new_job) = rx_new_job.recv() => { + debug!("Received NewExtendedMiningJob (Job ID: {:?})", new_job.job_id); + handle_result!( + tx_status.clone(), + Self::handle_new_extended_mining_job_(self_.clone(), new_job, tx_sv1_notify.clone()).await + ); + crate::upstream_sv2::upstream::IS_NEW_JOB_HANDLED + .store(true, std::sync::atomic::Ordering::SeqCst); + } + else => { + // One or both channels closed, indicating upstream disconnection or shutdown. + debug!("Upstream job channel(s) closed. Exiting job handler."); + break; + } + } + } + }); + + task_collector + .safe_lock(|c| c.push((handle.abort_handle(), "handle_upstream_job_handler".into()))) + .unwrap(); + } + /// Task handler that receives `DownstreamMessages` and dispatches them. /// /// This loop continuously receives messages from the `rx_sv1_downstream` channel. /// It matches on the `DownstreamMessages` variant and calls the appropriate /// handler function (`handle_submit_shares` or `handle_update_downstream_target`). fn handle_downstream_messages(self_: Arc>) { - let task_collector_handle_downstream = - self_.safe_lock(|b| b.task_collector.clone()).unwrap(); + let task_collector = self_.safe_lock(|b| b.task_collector.clone()).unwrap(); let (rx_sv1_downstream, tx_status) = self_ .safe_lock(|s| (s.rx_sv1_downstream.clone(), s.tx_status.clone())) .unwrap(); - let handle_downstream = tokio::task::spawn(async move { - loop { - let msg = handle_result!(tx_status, rx_sv1_downstream.clone().recv().await); - match msg { - DownstreamMessages::SubmitShares(share) => { - handle_result!( - tx_status, - Self::handle_submit_shares(self_.clone(), share).await - ); + let handle = tokio::task::spawn(async move { + loop { + match rx_sv1_downstream.recv().await { + Ok(msg) => { + let res = match msg { + DownstreamMessages::SubmitShares(share) => { + Self::handle_submit_shares(self_.clone(), share).await + } + DownstreamMessages::SetDownstreamTarget(new_target) => { + Self::handle_update_downstream_target(self_.clone(), new_target) + } + }; + handle_result!(tx_status.clone(), res); } - DownstreamMessages::SetDownstreamTarget(new_target) => { - handle_result!( - tx_status, - Self::handle_update_downstream_target(self_.clone(), new_target) - ); + Err(_) => { + debug!("Downstream channel closed. Exiting downstream handler."); + break; } - }; + } } }); - let _ = task_collector_handle_downstream.safe_lock(|a| { - a.push(( - handle_downstream.abort_handle(), - "handle_downstream_message".to_string(), - )) - }); + + task_collector + .safe_lock(|c| c.push((handle.abort_handle(), "handle_downstream_messages".into()))) + .unwrap(); } /// Receives a `SetDownstreamTarget` message and updates the downstream target for a specific @@ -252,9 +320,24 @@ impl Bridge { self_: Arc>, new_target: SetDownstreamTarget, ) -> ProxyResult<'static, ()> { - self_.safe_lock(|b| { - b.channel_factory - .update_target_for_channel(new_target.channel_id, new_target.new_target); + self_.safe_lock(|bridge| { + bridge + .upstream_channel_manager + .safe_lock(|upstream_channel_manager| { + let upstream_manager = upstream_channel_manager + .upstream_manager + .get_mut(&new_target.channel_id); + if let Some(upstream_manager) = upstream_manager { + let difficulty_config = upstream_manager + .downstream_manager + .difficulty_config + .get_mut(&new_target.connection_id); + if let Some(difficulty_config) = difficulty_config { + difficulty_config.target = new_target.new_target; + } + } + }) + .unwrap(); })?; Ok(()) } @@ -264,56 +347,32 @@ impl Bridge { self_: Arc>, share: SubmitShareWithChannelId, ) -> ProxyResult<'static, ()> { - let (tx_sv2_submit_shares_ext, target_mutex, tx_status) = self_.safe_lock(|s| { - ( - s.tx_sv2_submit_shares_ext.clone(), - s.target.clone(), - s.tx_status.clone(), - ) - })?; - let upstream_target: [u8; 32] = target_mutex.safe_lock(|t| t.clone())?.try_into()?; - let mut upstream_target: Target = upstream_target.into(); - self_.safe_lock(|s| s.channel_factory.set_target(&mut upstream_target))?; - - let sv2_submit = self_.safe_lock(|s| { - s.translate_submit(share.channel_id, share.share, share.version_rolling_mask) - })??; - let res = self_ - .safe_lock(|s| s.channel_factory.on_submit_shares_extended(sv2_submit)) - .map_err(|_| PoisonLock); - - match res { - Ok(Ok(OnNewShare::SendErrorDownstream(e))) => { - warn!( - "Submit share error {:?}", - std::str::from_utf8(&e.error_code.to_vec()[..]) - ); - } - Ok(Ok(OnNewShare::SendSubmitShareUpstream((share, _)))) => { - info!("SHARE MEETS UPSTREAM TARGET"); - match share { - Share::Extended(share) => { - tx_sv2_submit_shares_ext.send(share).await?; + let verdict = self_.safe_lock(|bridge| { + let verdict = bridge + .upstream_channel_manager + .safe_lock(|upstream_manager| { + if let Some(upstream_channel) = + upstream_manager.upstream_manager.get_mut(&share.channel_id) + { + return upstream_channel + .downstream_manager + .on_submit_share(share.clone()); } - // We are in an extended channel shares are extended - Share::Standard(_) => unreachable!(), - } - } - // We are in an extended channel this variant is group channle only - Ok(Ok(OnNewShare::RelaySubmitShareUpstream)) => unreachable!(), - Ok(Ok(OnNewShare::ShareMeetDownstreamTarget)) => { - debug!("SHARE MEETS DOWNSTREAM TARGET"); - } - // Proxy do not have JD capabilities - Ok(Ok(OnNewShare::ShareMeetBitcoinTarget(..))) => unreachable!(), - Ok(Err(e)) => error!("Error: {:?}", e), - Err(e) => { - let _ = tx_status - .send(status::Status { - state: status::State::BridgeShutdown(e), - }) - .await; - } + false + }) + .unwrap(); + verdict + })?; + info!("Share submission verdict: {verdict}"); + _ = share.verdict_sender.send(verdict).await; + let tx_sv2_submit_shares_ext = self_.safe_lock(|s| s.tx_sv2_submit_shares_ext.clone())?; + + if verdict { + let sv2_submit = self_.safe_lock(|s| { + s.translate_submit(share.channel_id, share.share, share.version_rolling_mask) + })??; + + tx_sv2_submit_shares_ext.send(sv2_submit).await?; } Ok(()) } @@ -331,28 +390,40 @@ impl Bridge { sv1_submit: Submit, version_rolling_mask: Option, ) -> ProxyResult<'static, SubmitSharesExtended<'static>> { - let last_version = self - .channel_factory - .last_valid_job_version() - .ok_or(Error::RolesSv2Logic(RolesLogicError::NoValidJob))?; - let version = match (sv1_submit.version_bits, version_rolling_mask) { - // regarding version masking see https://github.com/slushpool/stratumprotocol/blob/master/stratum-extensions.mediawiki#changes-in-request-miningsubmit - (Some(vb), Some(mask)) => (last_version & !mask.0) | (vb.0 & mask.0), - (None, None) => last_version, - _ => return Err(Error::V1Protocol(v1::error::Error::InvalidSubmission)), - }; - let mining_device_extranonce: Vec = sv1_submit.extra_nonce2.into(); - let extranonce2 = mining_device_extranonce; - Ok(SubmitSharesExtended { - channel_id, - // I put 0 below cause sequence_number is not what should be TODO - sequence_number: 0, - job_id: sv1_submit.job_id.parse::()?, - nonce: sv1_submit.nonce.0, - ntime: sv1_submit.time.0, - version, - extranonce: extranonce2.try_into()?, - }) + let job = self + .upstream_channel_manager + .safe_lock(|upstream_manager| { + let upstream_manager = upstream_manager.upstream_manager.get(&channel_id); + if let Some(upstream_channel) = upstream_manager { + if let Ok(job_id) = sv1_submit.job_id.parse::() { + return upstream_channel.downstream_manager.get_job(job_id); + } + } + None + }) + .unwrap(); + if let Some(job) = job { + let last_version = job.version; + let version = match (sv1_submit.version_bits, version_rolling_mask) { + // regarding version masking see https://github.com/slushpool/stratumprotocol/blob/master/stratum-extensions.mediawiki#changes-in-request-miningsubmit + (Some(vb), Some(mask)) => (last_version & !mask.0) | (vb.0 & mask.0), + (None, None) => last_version, + _ => return Err(Error::V1Protocol(v1::error::Error::InvalidSubmission)), + }; + let mining_device_extranonce: Vec = sv1_submit.extra_nonce2.into(); + let extranonce2 = mining_device_extranonce; + return Ok(SubmitSharesExtended { + channel_id, + // I put 0 below cause sequence_number is not what should be TODO + sequence_number: 0, + job_id: sv1_submit.job_id.parse::()?, + nonce: sv1_submit.nonce.0, + ntime: sv1_submit.time.0, + version, + extranonce: extranonce2.try_into()?, + }); + } + Err(Error::RolesSv2Logic(RolesLogicError::NoValidJob)) } /// Internal helper function to handle a received SV2 `SetNewPrevHash` message. @@ -368,95 +439,44 @@ impl Bridge { sv2_set_new_prev_hash: SetNewPrevHash<'static>, tx_sv1_notify: broadcast::Sender>, ) -> Result<(), Error<'static>> { - while !crate::upstream_sv2::upstream::IS_NEW_JOB_HANDLED - .load(std::sync::atomic::Ordering::SeqCst) - { - tokio::task::yield_now().await; - } - self_.safe_lock(|s| s.last_p_hash = Some(sv2_set_new_prev_hash.clone()))?; - - let on_new_prev_hash_res = self_.safe_lock(|s| { - s.channel_factory - .on_new_prev_hash(sv2_set_new_prev_hash.clone()) - })?; - on_new_prev_hash_res?; - - let mut future_jobs = self_.safe_lock(|s| { - let future_jobs = s.future_jobs.clone(); - s.future_jobs = vec![]; - future_jobs + // The handle_new_prev_hash_ by bridge shouldn't be doing + // any channel management as its job is just to translate + // and do nothing else. + // + // We are fetching the current active job from corresponding + // upstream channel, channel manager and creating its notification. + + // fetching the active job, for corresponding SetNewPrevHash message, which should already + // be populated in channel_manager. + let active_job = self_.safe_lock(|bridge| { + let value = bridge + .upstream_channel_manager + .safe_lock(|manager| { + let upstream_channel = manager + .upstream_manager + .get(&sv2_set_new_prev_hash.channel_id); + if let Some(upstream_channel) = upstream_channel { + return upstream_channel.downstream_manager.active_job.clone(); + } + None + }) + .unwrap(); + value })?; - let mut match_a_future_job = false; - while let Some(job) = future_jobs.pop() { - if job.job_id == sv2_set_new_prev_hash.job_id { - let j_id = job.job_id; - // Create the mining.notify to be sent to the Downstream. - let notify = crate::proxy::next_mining_notify::create_notify( - sv2_set_new_prev_hash.clone(), - job, - true, - ); + if let Some(active_job) = active_job { + // Sending the notify message to downstream. + let notify = crate::proxy::next_mining_notify::create_notify( + sv2_set_new_prev_hash.clone(), + active_job, + true, + ); - // Get the sender to send the mining.notify to the Downstream - tx_sv1_notify.send(notify.clone())?; - match_a_future_job = true; - self_.safe_lock(|s| { - s.last_notify = Some(notify); - s.last_job_id = j_id; - })?; - break; - } - } - if !match_a_future_job { - debug!("No future jobs for {:?}", sv2_set_new_prev_hash); + // Get the sender to send the mining.notify to the Downstream + tx_sv1_notify.send(notify.clone())?; } - Ok(()) - } - /// Task handler that receives SV2 `SetNewPrevHash` messages from the upstream. - /// - /// This loop continuously receives `SetNewPrevHash` messages. It calls the - /// internal `handle_new_prev_hash_` helper function to process each message. - fn handle_new_prev_hash(self_: Arc>) { - let task_collector_handle_new_prev_hash = - self_.safe_lock(|b| b.task_collector.clone()).unwrap(); - let (tx_sv1_notify, rx_sv2_set_new_prev_hash, tx_status) = self_ - .safe_lock(|s| { - ( - s.tx_sv1_notify.clone(), - s.rx_sv2_set_new_prev_hash.clone(), - s.tx_status.clone(), - ) - }) - .unwrap(); - debug!("Starting handle_new_prev_hash task"); - let handle_new_prev_hash = tokio::task::spawn(async move { - loop { - // Receive `SetNewPrevHash` from `Upstream` - let sv2_set_new_prev_hash: SetNewPrevHash = - handle_result!(tx_status, rx_sv2_set_new_prev_hash.clone().recv().await); - debug!( - "handle_new_prev_hash job_id: {:?}", - &sv2_set_new_prev_hash.job_id - ); - handle_result!( - tx_status.clone(), - Self::handle_new_prev_hash_( - self_.clone(), - sv2_set_new_prev_hash, - tx_sv1_notify.clone(), - ) - .await - ) - } - }); - let _ = task_collector_handle_new_prev_hash.safe_lock(|a| { - a.push(( - handle_new_prev_hash.abort_handle(), - "handle_new_prev_hash".to_string(), - )) - }); + Ok(()) } /// Internal helper function to handle a received SV2 `NewExtendedMiningJob` message. @@ -472,95 +492,47 @@ impl Bridge { sv2_new_extended_mining_job: NewExtendedMiningJob<'static>, tx_sv1_notify: broadcast::Sender>, ) -> Result<(), Error<'static>> { - // convert to non segwit jobs so we dont have to depend if miner's support segwit or not - self_.safe_lock(|s| { - s.channel_factory - .on_new_extended_mining_job(sv2_new_extended_mining_job.as_static().clone()) - })??; - - // If future_job=true, this job is meant for a future SetNewPrevHash that the proxy - // has yet to receive. Insert this new job into the job_mapper . - if sv2_new_extended_mining_job.is_future() { - self_.safe_lock(|s| s.future_jobs.push(sv2_new_extended_mining_job.clone()))?; - Ok(()) + // The handle_new_extended_mining_job_ by bridge shouldn't be doing + // any channel management as its job is just to translate + // and do nothing else. + // + // We are fetching the current previous block hash from corresponding + // upstream channel, channel manager and creating its notification. - // If future_job=false, this job is meant for the current SetNewPrevHash. - } else { - let last_p_hash_option = self_.safe_lock(|s| s.last_p_hash.clone())?; + if sv2_new_extended_mining_job.is_future() { + return Ok(()); + } - // last_p_hash is an Option so we need to map to the correct error type - // to be handled - let last_p_hash = last_p_hash_option.ok_or(Error::RolesSv2Logic( - RolesLogicError::JobIsNotFutureButPrevHashNotPresent, - ))?; + // fetching the active job, for corresponding SetNewPrevHash message, which should already + // be populated in channel_manager. + let prev_block_hash = self_.safe_lock(|bridge| { + let value = bridge + .upstream_channel_manager + .safe_lock(|manager| { + let upstream_channel = manager + .upstream_manager + .get(&sv2_new_extended_mining_job.channel_id); + if let Some(upstream_channel) = upstream_channel { + return upstream_channel.downstream_manager.prev_block_hash.clone(); + } + None + }) + .unwrap(); + value + })?; - let j_id = sv2_new_extended_mining_job.job_id; - // Create the mining.notify to be sent to the Downstream. - // clean_jobs must be false because it's not a NewPrevHash template + if let Some(prev_block_hash) = prev_block_hash { + // Sending the notify message to downstream. let notify = crate::proxy::next_mining_notify::create_notify( - last_p_hash, - sv2_new_extended_mining_job.clone(), - false, + prev_block_hash, + sv2_new_extended_mining_job, + true, ); // Get the sender to send the mining.notify to the Downstream tx_sv1_notify.send(notify.clone())?; - self_.safe_lock(|s| { - s.last_notify = Some(notify); - s.last_job_id = j_id; - })?; - Ok(()) } - } - /// Task handler that receives SV2 `NewExtendedMiningJob` messages from the upstream. - /// - /// This loop continuously receives `NewExtendedMiningJob` messages. It calls the - /// internal `handle_new_extended_mining_job_` helper function to process each message. - /// After processing, it signals that a new job has been handled (used for synchronization - /// with the `handle_new_prev_hash` task). - fn handle_new_extended_mining_job(self_: Arc>) { - let task_collector_new_extended_mining_job = - self_.safe_lock(|b| b.task_collector.clone()).unwrap(); - let (tx_sv1_notify, rx_sv2_new_ext_mining_job, tx_status) = self_ - .safe_lock(|s| { - ( - s.tx_sv1_notify.clone(), - s.rx_sv2_new_ext_mining_job.clone(), - s.tx_status.clone(), - ) - }) - .unwrap(); - debug!("Starting handle_new_extended_mining_job task"); - let handle_new_extended_mining_job = tokio::task::spawn(async move { - loop { - // Receive `NewExtendedMiningJob` from `Upstream` - let sv2_new_extended_mining_job: NewExtendedMiningJob = handle_result!( - tx_status.clone(), - rx_sv2_new_ext_mining_job.clone().recv().await - ); - debug!( - "handle_new_extended_mining_job job_id: {:?}", - &sv2_new_extended_mining_job.job_id - ); - handle_result!( - tx_status, - Self::handle_new_extended_mining_job_( - self_.clone(), - sv2_new_extended_mining_job, - tx_sv1_notify.clone(), - ) - .await - ); - crate::upstream_sv2::upstream::IS_NEW_JOB_HANDLED - .store(true, std::sync::atomic::Ordering::SeqCst); - } - }); - let _ = task_collector_new_extended_mining_job.safe_lock(|a| { - a.push(( - handle_new_extended_mining_job.abort_handle(), - "handle_new_extended_mining_job".to_string(), - )) - }); + Ok(()) } } @@ -571,168 +543,15 @@ impl Bridge { /// channel ID assigned to the connection, the initial job notification to send, /// and the extranonce and target specific to this channel. pub struct OpenSv1Downstream { - /// The unique ID assigned to this downstream channel by the channel factory. + /// The unique ID assigned to this upstream channel by the channel factory. pub channel_id: u32, + /// This is use to pin point a single downstream channel. + pub connection_id: Sv1ChannelId, /// The most recent `mining.notify` message to send to the new client immediately /// upon connection to provide them with a job. pub last_notify: Option>, /// The extranonce prefix assigned to this channel. pub extranonce: Vec, - /// The mining target assigned to this channel - pub target: Arc>>, /// The size of the extranonce2 field expected from the miner for this channel. pub extranonce2_len: u16, } - -#[cfg(test)] -mod test { - use super::*; - use async_channel::bounded; - use stratum_common::bitcoin::{absolute::LockTime, consensus, transaction::Version}; - - pub mod test_utils { - use super::*; - - #[allow(dead_code)] - pub struct BridgeInterface { - pub tx_sv1_submit: Sender, - pub rx_sv2_submit_shares_ext: Receiver>, - pub tx_sv2_set_new_prev_hash: Sender>, - pub tx_sv2_new_ext_mining_job: Sender>, - pub rx_sv1_notify: broadcast::Receiver>, - } - - pub fn create_bridge( - extranonces: ExtendedExtranonce, - ) -> (Arc>, BridgeInterface) { - let (tx_sv1_submit, rx_sv1_submit) = bounded(1); - let (tx_sv2_submit_shares_ext, rx_sv2_submit_shares_ext) = bounded(1); - let (tx_sv2_set_new_prev_hash, rx_sv2_set_new_prev_hash) = bounded(1); - let (tx_sv2_new_ext_mining_job, rx_sv2_new_ext_mining_job) = bounded(1); - let (tx_sv1_notify, rx_sv1_notify) = broadcast::channel(1); - let (tx_status, _rx_status) = bounded(1); - let upstream_target = vec![ - 0, 0, 0, 0, 255, 255, 255, 255, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, - 0, 0, 0, 0, 0, 0, 0, - ]; - let interface = BridgeInterface { - tx_sv1_submit, - rx_sv2_submit_shares_ext, - tx_sv2_set_new_prev_hash, - tx_sv2_new_ext_mining_job, - rx_sv1_notify, - }; - - let task_collector = Arc::new(Mutex::new(vec![])); - let b = Bridge::new( - rx_sv1_submit, - tx_sv2_submit_shares_ext, - rx_sv2_set_new_prev_hash, - rx_sv2_new_ext_mining_job, - tx_sv1_notify, - status::Sender::Bridge(tx_status), - extranonces, - Arc::new(Mutex::new(upstream_target)), - 1, - task_collector, - ); - (b, interface) - } - - pub fn create_sv1_submit(job_id: u32) -> Submit<'static> { - Submit { - user_name: "test_user".to_string(), - job_id: job_id.to_string(), - extra_nonce2: v1::utils::Extranonce::try_from([0; 32].to_vec()).unwrap(), - time: v1::utils::HexU32Be(1), - nonce: v1::utils::HexU32Be(1), - version_bits: None, - id: 0, - } - } - } - - #[test] - fn test_version_bits_insert() { - use stratum_common::{ - bitcoin, - bitcoin::{blockdata::witness::Witness, hashes::Hash}, - }; - - let extranonces = ExtendedExtranonce::new(0..6, 6..8, 8..16, None) - .expect("Failed to create ExtendedExtranonce with valid ranges"); - let (bridge, _) = test_utils::create_bridge(extranonces); - bridge - .safe_lock(|bridge| { - let channel_id = 1; - let out_id = bitcoin::hashes::sha256d::Hash::from_slice(&[ - 0_u8, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, - 0, 0, 0, 0, 0, 0, 0, - ]) - .unwrap(); - let p_out = bitcoin::OutPoint { - txid: bitcoin::Txid::from_raw_hash(out_id), - vout: 0xffff_ffff, - }; - let in_ = bitcoin::TxIn { - previous_output: p_out, - script_sig: vec![89_u8; 16].into(), - sequence: bitcoin::Sequence(0), - witness: Witness::from(vec![] as Vec>), - }; - let tx = bitcoin::Transaction { - version: Version::ONE, - lock_time: LockTime::from_consensus(0), - input: vec![in_], - output: vec![], - }; - let tx = consensus::serialize(&tx); - let _down = bridge - .channel_factory - .add_standard_channel(0, 10_000_000_000.0, true, 1) - .unwrap(); - let prev_hash = SetNewPrevHash { - channel_id, - job_id: 0, - prev_hash: [ - 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, 3, - 3, 3, 3, 3, 3, 3, 3, - ] - .into(), - min_ntime: 989898, - nbits: 9, - }; - bridge.channel_factory.on_new_prev_hash(prev_hash).unwrap(); - let now = std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .unwrap() - .as_secs() as u32; - let new_mining_job = NewExtendedMiningJob { - channel_id, - job_id: 0, - min_ntime: binary_sv2::Sv2Option::new(Some(now)), - version: 0b0000_0000_0000_0000, - version_rolling_allowed: false, - merkle_path: vec![].into(), - coinbase_tx_prefix: tx[0..42].to_vec().try_into().unwrap(), - coinbase_tx_suffix: tx[58..].to_vec().try_into().unwrap(), - }; - bridge - .channel_factory - .on_new_extended_mining_job(new_mining_job.clone()) - .unwrap(); - - // pass sv1_submit into Bridge::translate_submit - let sv1_submit = test_utils::create_sv1_submit(0); - let sv2_message = bridge - .translate_submit(channel_id, sv1_submit, None) - .unwrap(); - // assert sv2 message equals sv1 with version bits added - assert_eq!( - new_mining_job.version, sv2_message.version, - "Version bits were not inserted for non version rolling sv1 message" - ); - }) - .unwrap(); - } -} diff --git a/roles/translator/src/lib/upstream_sv2/diff_management.rs b/roles/translator/src/lib/upstream_sv2/diff_management.rs index 7cdd585e25..8d8878e724 100644 --- a/roles/translator/src/lib/upstream_sv2/diff_management.rs +++ b/roles/translator/src/lib/upstream_sv2/diff_management.rs @@ -8,6 +8,8 @@ //! `UpdateChannel` messages to the upstream server //! based on configured nominal hashrate changes. +use crate::config::UpstreamDifficultyConfig; + use super::Upstream; use super::super::{ @@ -15,52 +17,62 @@ use super::super::{ upstream_sv2::{EitherFrame, Message, StdFrame}, }; use binary_sv2::u256_from_int; -use roles_logic_sv2::{ - mining_sv2::UpdateChannel, parsers::Mining, utils::Mutex, Error as RolesLogicError, -}; +use roles_logic_sv2::{mining_sv2::UpdateChannel, parsers::Mining, utils::Mutex}; use std::{sync::Arc, time::Duration}; impl Upstream { /// Attempts to update the upstream channel's nominal hashrate if the configured /// update interval has elapsed or if the nominal hashrate has changed pub(super) async fn try_update_hashrate(self_: Arc>) -> ProxyResult<'static, ()> { - let (channel_id_option, diff_mgmt, tx_frame, last_sent_hashrate) = - self_.safe_lock(|u| { - ( - u.channel_id, - u.difficulty_config.clone(), - u.connection.sender.clone(), - u.last_sent_hashrate, - ) - })?; - - let channel_id = channel_id_option.ok_or(super::super::error::Error::RolesSv2Logic( - RolesLogicError::NotFoundChannelId, - ))?; - - let (timeout, new_hashrate) = diff_mgmt - .safe_lock(|d| (d.channel_diff_update_interval, d.channel_nominal_hashrate))?; + let tx_frame = self_.safe_lock(|u| u.connection.sender.clone())?; + let result = self_.safe_lock(|upstream| { + let result: Vec<(u32, UpstreamDifficultyConfig, f32)> = upstream + .upstream_channel_manager + .safe_lock(|upstream_manager| { + let result = upstream_manager + .upstream_manager + .iter() + .map(|(k, v)| (*k, v.upstream_difficulty.clone(), v.last_sent_hashrate)) + .collect(); + result + }) + .unwrap(); + result + })?; - let has_changed = Some(new_hashrate) != last_sent_hashrate; + for (channel_id, diff_mgmt, last_sent_hashrate) in result { + let new_hashrate = diff_mgmt.channel_nominal_hashrate; - if has_changed { - // Send UpdateChannel only if hashrate actually changed - let update_channel = UpdateChannel { - channel_id, - nominal_hash_rate: new_hashrate, - maximum_target: u256_from_int(u64::MAX), - }; - let message = Message::Mining(Mining::UpdateChannel(update_channel)); - let either_frame: StdFrame = message.try_into()?; - let frame: EitherFrame = either_frame.into(); + let has_changed = new_hashrate != last_sent_hashrate; - tx_frame.send(frame).await?; + if has_changed { + // Send UpdateChannel only if hashrate actually changed + let update_channel = UpdateChannel { + channel_id, + nominal_hash_rate: new_hashrate, + maximum_target: u256_from_int(u64::MAX), + }; + let message = Message::Mining(Mining::UpdateChannel(update_channel)); + let either_frame: StdFrame = message.try_into()?; + let frame: EitherFrame = either_frame.into(); - self_.safe_lock(|u| u.last_sent_hashrate = Some(new_hashrate))?; + tx_frame.send(frame).await?; + self_.safe_lock(|upstream| { + _ = upstream + .upstream_channel_manager + .safe_lock(|upstream_manager| { + if let Some(upstream_channel) = + upstream_manager.upstream_manager.get_mut(&channel_id) + { + upstream_channel.last_sent_hashrate = new_hashrate; + } + }); + })?; + } } // Always sleep, regardless of update - tokio::time::sleep(Duration::from_secs(timeout as u64)).await; + tokio::time::sleep(Duration::from_secs(60_u64)).await; Ok(()) } } diff --git a/roles/translator/src/lib/upstream_sv2/message_handler.rs b/roles/translator/src/lib/upstream_sv2/message_handler.rs new file mode 100644 index 0000000000..efed80b71c --- /dev/null +++ b/roles/translator/src/lib/upstream_sv2/message_handler.rs @@ -0,0 +1,366 @@ +use roles_logic_sv2::{ + common_messages_sv2::Protocol, + common_properties::{IsMiningUpstream, IsUpstream}, + handlers::mining::{ParseMiningMessagesFromUpstream, SendTo, SupportedChannelTypes}, + mining_sv2::{NewExtendedMiningJob, SetNewPrevHash}, + parsers::Mining, + Error as RolesLogicError, +}; +use tracing::info; + +use crate::{ + channel_manager::{ChannelManager, UpstreamChannel}, + config::UpstreamDifficultyConfig, + downstream_sv1::Downstream, + upstream_sv2::upstream::IS_NEW_JOB_HANDLED, +}; + +use tracing::{debug, error, warn}; + +use roles_logic_sv2::mining_sv2::SetGroupChannel; + +use super::upstream::Upstream; + +// Can be removed? +impl IsUpstream for Upstream { + fn get_version(&self) -> u16 { + todo!() + } + + fn get_flags(&self) -> u32 { + todo!() + } + + fn get_supported_protocols(&self) -> Vec { + todo!() + } + + fn get_id(&self) -> u32 { + todo!() + } + + fn get_mapper(&mut self) -> Option<&mut roles_logic_sv2::common_properties::RequestIdMapper> { + todo!() + } +} + +// Can be removed? +impl IsMiningUpstream for Upstream { + fn total_hash_rate(&self) -> u64 { + todo!() + } + + fn add_hash_rate(&mut self, _to_add: u64) { + todo!() + } + + fn get_opened_channels( + &mut self, + ) -> &mut Vec { + todo!() + } + + fn update_channels(&mut self, _c: roles_logic_sv2::common_properties::UpstreamChannel) { + todo!() + } +} + +/// Connection-wide SV2 Upstream role messages parser implemented by a downstream ("downstream" +/// here is relative to the SV2 Upstream role and is represented by this `Upstream` struct). +impl ParseMiningMessagesFromUpstream for Upstream { + /// Returns the type of channel used between this proxy and the SV2 Upstream. + /// For a Translator Proxy, this is always `Extended`. + fn get_channel_type(&self) -> SupportedChannelTypes { + SupportedChannelTypes::Extended + } + + /// Indicates whether work selection is enabled for this upstream connection. + /// For a Translator Proxy, work selection is handled by the upstream pool, + /// so this method always returns `false`. + fn is_work_selection_enabled(&self) -> bool { + false + } + + /// The SV2 `OpenStandardMiningChannelSuccess` message is NOT handled because it is NOT used + /// for the Translator Proxy as only `Extended` channels are used between the SV1/SV2 Translator + /// Proxy and the SV2 Upstream role. + fn handle_open_standard_mining_channel_success( + &mut self, + _m: roles_logic_sv2::mining_sv2::OpenStandardMiningChannelSuccess, + ) -> Result, RolesLogicError> { + panic!("Standard Mining Channels are not used in Translator Proxy") + } + + /// Handles the SV2 `OpenExtendedMiningChannelSuccess` message. + /// + /// This message is received after requesting to open an extended mining channel. + /// It provides the assigned `channel_id`, the extranonce prefix, the initial + /// mining `target`, and the expected `extranonce_size`. It stores the `channel_id` and + /// `extranonce_prefix`, updates the shared `target`, and prepares the extranonce + /// information (including calculating the size for the TProxy's added extranonce1) to be + /// sent to the Downstream handler for use with SV1 clients. + /// + /// Returns `Ok(SendTo::None(Some(Mining::OpenExtendedMiningChannelSuccess)))` + /// to indicate that the message has been handled internally and should be + /// forwarded to the Bridge. + fn handle_open_extended_mining_channel_success( + &mut self, + m: roles_logic_sv2::mining_sv2::OpenExtendedMiningChannelSuccess, + ) -> Result, RolesLogicError> { + info!( + "Received OpenExtendedMiningChannelSuccess with request id: {} and channel id: {}", + m.request_id, m.channel_id + ); + + let min_extranonce_size = self + .upstream_channel_manager + .safe_lock(|u| u.min_extranonce_size)?; + + debug!("OpenExtendedMiningChannelSuccess: {:?}", m); + let tproxy_e1_len = super::super::utils::proxy_extranonce1_len( + m.extranonce_size as usize, + min_extranonce_size.into(), + ) as u16; + if min_extranonce_size + tproxy_e1_len < m.extranonce_size { + return Err(RolesLogicError::InvalidExtranonceSize( + min_extranonce_size, + m.extranonce_size, + )); + } + + info!("Up: Successfully Opened Extended Mining Channel"); + + self.upstream_channel_manager.safe_lock(|e| { + info!("Updating upstream channel manager state with new upstream connection"); + + e.channel_ids.insert(m.channel_id); + let downstream_channel_manager = ChannelManager::new( + m.extranonce_prefix.clone().into(), + m.extranonce_prefix.to_vec().len(), + m.extranonce_size as usize, + min_extranonce_size as usize, + e.shares_per_minute, + m.channel_id, + ); + + let upstream_difficulty = UpstreamDifficultyConfig { + channel_nominal_hashrate: 0.0, + channel_diff_update_interval: e.update_interval, + should_aggregate: true, + timestamp_of_last_update: 0, + }; + + let upstream_channel = UpstreamChannel::new( + downstream_channel_manager, + upstream_difficulty.channel_nominal_hashrate, + upstream_difficulty, + m.target.clone().into(), + ); + e.upstream_manager.insert(m.channel_id, upstream_channel); + e.request_id_to_channel_id + .insert(m.request_id, m.channel_id); + })?; + + let m = Mining::OpenExtendedMiningChannelSuccess(m.into_static()); + Ok(SendTo::None(Some(m))) + } + + /// Handles the SV2 `OpenExtendedMiningChannelError` message (TODO). + fn handle_open_mining_channel_error( + &mut self, + m: roles_logic_sv2::mining_sv2::OpenMiningChannelError, + ) -> Result, RolesLogicError> { + error!( + "Received OpenExtendedMiningChannelError with error code {}", + std::str::from_utf8(m.error_code.as_ref()).unwrap_or("unknown error code") + ); + Ok(SendTo::None(Some(Mining::OpenMiningChannelError( + m.as_static(), + )))) + } + /// Handles the SV2 `UpdateChannelError` message (TODO). + fn handle_update_channel_error( + &mut self, + m: roles_logic_sv2::mining_sv2::UpdateChannelError, + ) -> Result, RolesLogicError> { + error!( + "Received UpdateChannelError with error code {}", + std::str::from_utf8(m.error_code.as_ref()).unwrap_or("unknown error code") + ); + Ok(SendTo::None(Some(Mining::UpdateChannelError( + m.as_static(), + )))) + } + + /// Handles the SV2 `CloseChannel` message (TODO). + fn handle_close_channel( + &mut self, + m: roles_logic_sv2::mining_sv2::CloseChannel, + ) -> Result, RolesLogicError> { + info!("Received CloseChannel for channel id: {}", m.channel_id); + + self.upstream_channel_manager.safe_lock(|u| { + // Todo improve this. + u.remove(m.channel_id); + })?; + + Ok(SendTo::None(Some(Mining::CloseChannel(m.as_static())))) + } + + /// Handles the SV2 `SetExtranoncePrefix` message (TODO). + fn handle_set_extranonce_prefix( + &mut self, + _: roles_logic_sv2::mining_sv2::SetExtranoncePrefix, + ) -> Result, RolesLogicError> { + todo!() + } + + /// Handles the SV2 `SubmitSharesSuccess` message. + fn handle_submit_shares_success( + &mut self, + m: roles_logic_sv2::mining_sv2::SubmitSharesSuccess, + ) -> Result, RolesLogicError> { + info!("Received SubmitSharesSuccess"); + debug!("SubmitSharesSuccess: {:?}", m); + Ok(SendTo::None(None)) + } + + /// Handles the SV2 `SubmitSharesError` message. + fn handle_submit_shares_error( + &mut self, + m: roles_logic_sv2::mining_sv2::SubmitSharesError, + ) -> Result, RolesLogicError> { + error!( + "Received SubmitSharesError with error code {}", + std::str::from_utf8(m.error_code.as_ref()).unwrap_or("unknown error code") + ); + Ok(SendTo::None(None)) + } + + /// The SV2 `NewMiningJob` message is NOT handled because it is NOT used for the Translator + /// Proxy as only `Extended` channels are used between the SV1/SV2 Translator Proxy and the SV2 + /// Upstream role. + fn handle_new_mining_job( + &mut self, + _m: roles_logic_sv2::mining_sv2::NewMiningJob, + ) -> Result, RolesLogicError> { + panic!("Standard Mining Channels are not used in Translator Proxy") + } + + /// Handles the SV2 `NewExtendedMiningJob` message which is used (along with the SV2 + /// `SetNewPrevHash` message) to later create a SV1 `mining.notify` for the Downstream + /// role. + fn handle_new_extended_mining_job( + &mut self, + m: NewExtendedMiningJob, + ) -> Result, RolesLogicError> { + info!( + "Received new extended mining job for channel id: {} with job id: {} is_future: {}", + m.channel_id, + m.job_id, + m.is_future() + ); + debug!("NewExtendedMiningJob: {:?}", m); + + self.upstream_channel_manager.safe_lock(|u| { + let channel_manager = u.upstream_manager.get_mut(&m.channel_id); + if let Some(channel_manager) = channel_manager { + channel_manager + .downstream_manager + .on_new_extended_job(m.clone().as_static()); + } + })?; + + if self.is_work_selection_enabled() { + Ok(SendTo::None(None)) + } else { + IS_NEW_JOB_HANDLED.store(false, std::sync::atomic::Ordering::SeqCst); + if !m.version_rolling_allowed { + warn!("VERSION ROLLING NOT ALLOWED IS A TODO"); + // todo!() + } + + let message = Mining::NewExtendedMiningJob(m.into_static()); + + Ok(SendTo::None(Some(message))) + } + } + + /// Handles the SV2 `SetNewPrevHash` message which is used (along with the SV2 + /// `NewExtendedMiningJob` message) to later create a SV1 `mining.notify` for the Downstream + /// role. + fn handle_set_new_prev_hash( + &mut self, + m: SetNewPrevHash, + ) -> Result, RolesLogicError> { + info!( + "Received SetNewPrevHash channel id: {}, job id: {}", + m.channel_id, m.job_id + ); + + self.upstream_channel_manager.safe_lock(|u| { + let channel_manager = u.upstream_manager.get_mut(&m.channel_id); + if let Some(channel_manager) = channel_manager { + channel_manager + .downstream_manager + .on_new_prev_hash(m.clone().as_static()); + } + })?; + + debug!("SetNewPrevHash: {:?}", m); + if self.is_work_selection_enabled() { + Ok(SendTo::None(None)) + } else { + let message = Mining::SetNewPrevHash(m.into_static()); + Ok(SendTo::None(Some(message))) + } + } + + /// Handles the SV2 `SetCustomMiningJobSuccess` message (TODO). + fn handle_set_custom_mining_job_success( + &mut self, + m: roles_logic_sv2::mining_sv2::SetCustomMiningJobSuccess, + ) -> Result, RolesLogicError> { + info!( + "Received SetCustomMiningJobSuccess for channel id: {} for job id: {}", + m.channel_id, m.job_id + ); + debug!("SetCustomMiningJobSuccess: {:?}", m); + debug!("Tproxy will never receive this message, and if it does, kindly ignore"); + Ok(SendTo::None(None)) + } + + /// Handles the SV2 `SetCustomMiningJobError` message (TODO). + fn handle_set_custom_mining_job_error( + &mut self, + _m: roles_logic_sv2::mining_sv2::SetCustomMiningJobError, + ) -> Result, RolesLogicError> { + unimplemented!() + } + + /// Handles the SV2 `SetTarget` message which updates the Downstream role(s) target + /// difficulty via the SV1 `mining.set_difficulty` message. + fn handle_set_target( + &mut self, + m: roles_logic_sv2::mining_sv2::SetTarget, + ) -> Result, RolesLogicError> { + info!("Received SetTarget for channel id: {}", m.channel_id); + debug!("SetTarget: {:?}", m); + let m = m.into_static(); + self.upstream_channel_manager + .safe_lock(|upstream_channel| { + let upstream_channel = upstream_channel.upstream_manager.get_mut(&m.channel_id); + if let Some(upstream_channel) = upstream_channel { + upstream_channel.target = m.maximum_target.into(); + } + })?; + Ok(SendTo::None(None)) + } + + fn handle_set_group_channel( + &mut self, + _m: SetGroupChannel, + ) -> Result, RolesLogicError> { + todo!() + } +} diff --git a/roles/translator/src/lib/upstream_sv2/mod.rs b/roles/translator/src/lib/upstream_sv2/mod.rs index 64f24acd32..d7ca4d17ae 100644 --- a/roles/translator/src/lib/upstream_sv2/mod.rs +++ b/roles/translator/src/lib/upstream_sv2/mod.rs @@ -12,6 +12,8 @@ use codec_sv2::{StandardEitherFrame, StandardSv2Frame}; use roles_logic_sv2::parsers::AnyMessage; pub mod diff_management; +pub mod message_handler; +pub mod setup_connection; pub mod upstream; pub mod upstream_connection; pub use upstream::Upstream; @@ -20,12 +22,3 @@ pub use upstream_connection::UpstreamConnection; pub type Message = AnyMessage<'static>; pub type StdFrame = StandardSv2Frame; pub type EitherFrame = StandardEitherFrame; - -/// Represents the state or parameters negotiated during an SV2 Setup Connection message. -#[derive(Clone, Copy, Debug)] -pub struct Sv2MiningConnection { - _version: u16, - _setup_connection_flags: u32, - #[allow(dead_code)] - setup_connection_success_flags: u32, -} diff --git a/roles/translator/src/lib/upstream_sv2/setup_connection.rs b/roles/translator/src/lib/upstream_sv2/setup_connection.rs new file mode 100644 index 0000000000..e4e1dd32fd --- /dev/null +++ b/roles/translator/src/lib/upstream_sv2/setup_connection.rs @@ -0,0 +1,85 @@ +use roles_logic_sv2::{ + common_messages_sv2::{Protocol, SetupConnection}, + handlers::common::{ParseCommonMessagesFromUpstream, SendTo as SendToCommon}, + Error as RolesLogicError, +}; +use tracing::info; + +use crate::error::ProxyResult; + +use roles_logic_sv2::common_messages_sv2::Reconnect; + +use super::upstream::Upstream; + +impl Upstream { + // Creates the initial `SetupConnection` message for the SV2 handshake. + // + // This message contains information about the proxy acting as a mining device, + // including supported protocol versions, flags, and hardcoded endpoint details. + // + // TODO: The Mining Device information is currently hardcoded. It should ideally + // be configurable or derived from the downstream connections. + #[allow(clippy::result_large_err)] + pub fn get_setup_connection_message( + min_version: u16, + max_version: u16, + is_work_selection_enabled: bool, + ) -> ProxyResult<'static, SetupConnection<'static>> { + let endpoint_host = "0.0.0.0".to_string().into_bytes().try_into()?; + let vendor = String::new().try_into()?; + let hardware_version = String::new().try_into()?; + let firmware = String::new().try_into()?; + let device_id = String::new().try_into()?; + let flags = match is_work_selection_enabled { + false => 0b0000_0000_0000_0000_0000_0000_0000_0100, + true => 0b0000_0000_0000_0000_0000_0000_0000_0110, + }; + Ok(SetupConnection { + protocol: Protocol::MiningProtocol, + min_version, + max_version, + flags, + endpoint_host, + endpoint_port: 50, + vendor, + hardware_version, + firmware, + device_id, + }) + } +} + +impl ParseCommonMessagesFromUpstream for Upstream { + // Handles the SV2 `SetupConnectionSuccess` message received from the upstream. + // + // Returns `Ok(SendToCommon::None(None))` as this message is handled internally + // and does not require a direct response or forwarding. + fn handle_setup_connection_success( + &mut self, + m: roles_logic_sv2::common_messages_sv2::SetupConnectionSuccess, + ) -> Result { + info!( + "Received `SetupConnectionSuccess`: version={}, flags={:b}", + m.used_version, m.flags + ); + Ok(SendToCommon::None(None)) + } + + fn handle_setup_connection_error( + &mut self, + _: roles_logic_sv2::common_messages_sv2::SetupConnectionError, + ) -> Result { + todo!() + } + + fn handle_channel_endpoint_changed( + &mut self, + _: roles_logic_sv2::common_messages_sv2::ChannelEndpointChanged, + ) -> Result { + todo!() + } + + fn handle_reconnect(&mut self, _m: Reconnect) -> Result { + todo!() + } +} diff --git a/roles/translator/src/lib/upstream_sv2/upstream.rs b/roles/translator/src/lib/upstream_sv2/upstream.rs index 841daf05e5..c00409387e 100644 --- a/roles/translator/src/lib/upstream_sv2/upstream.rs +++ b/roles/translator/src/lib/upstream_sv2/upstream.rs @@ -18,14 +18,14 @@ //! `ParseCommonMessagesFromUpstream`, `ParseMiningMessagesFromUpstream`). use crate::{ - config::UpstreamDifficultyConfig, - downstream_sv1::Downstream, + channel_manager::UpstreamChannelManager, error::{ - Error::{CodecNoise, InvalidExtranonce, PoisonLock, UpstreamIncoming}, + Error::{CodecNoise, PoisonLock, UpstreamIncoming}, ProxyResult, }, status, upstream_sv2::{EitherFrame, Message, StdFrame, UpstreamConnection}, + OpenConnection, }; use async_channel::{Receiver, Sender}; use binary_sv2::u256_from_int; @@ -34,19 +34,15 @@ use error_handling::handle_result; use key_utils::Secp256k1PublicKey; use network_helpers_sv2::noise_connection::Connection; use roles_logic_sv2::{ - common_messages_sv2::{Protocol, SetupConnection}, - common_properties::{IsMiningUpstream, IsUpstream}, handlers::{ - common::{ParseCommonMessagesFromUpstream, SendTo as SendToCommon}, + common::ParseCommonMessagesFromUpstream, mining::{ParseMiningMessagesFromUpstream, SendTo}, }, mining_sv2::{ - ExtendedExtranonce, Extranonce, NewExtendedMiningJob, OpenExtendedMiningChannel, - SetNewPrevHash, SubmitSharesExtended, + NewExtendedMiningJob, OpenExtendedMiningChannel, SetNewPrevHash, SubmitSharesExtended, }, parsers::Mining, utils::Mutex, - Error as RolesLogicError, Error::NoUpstreamsConnected, }; use std::{ @@ -58,28 +54,12 @@ use tokio::{ task::AbortHandle, time::{sleep, Duration}, }; -use tracing::{debug, error, info, warn}; - -use roles_logic_sv2::{ - common_messages_sv2::Reconnect, handlers::mining::SupportedChannelTypes, - mining_sv2::SetGroupChannel, -}; -use stratum_common::bitcoin::BlockHash; +use tracing::{error, info, warn}; /// Atomic boolean flag used for synchronization between receiving a new job /// and handling a new previous hash. Indicates whether a `NewExtendedMiningJob` /// has been fully processed. pub static IS_NEW_JOB_HANDLED: AtomicBool = AtomicBool::new(true); -/// Represents the currently active `prevhash` of the mining job being worked on OR being submitted -/// from the Downstream role. -#[derive(Debug, Clone)] -#[allow(dead_code)] -struct PrevHash { - /// `prevhash` of mining job. - prev_hash: BlockHash, - /// `nBits` encoded difficulty target. - nbits: u32, -} /// Represents a connection to a single SV2 Upstream role. /// @@ -88,15 +68,6 @@ struct PrevHash { /// templates, and managing the SV2 protocol handshake and channel lifecycle. #[derive(Debug, Clone)] pub struct Upstream { - /// Newly assigned identifier of the channel, stable for the whole lifetime of the connection, - /// e.g. it is used for broadcasting new jobs by the `NewExtendedMiningJob` message. - pub(super) channel_id: Option, - /// Identifier of the job as provided by the `NewExtendedMiningJob` message. - job_id: Option, - /// Identifier of the job as provided by the ` SetCustomMiningJobSucces` message - last_job_id: Option, - /// Bytes used as implicit first part of `extranonce`. - extranonce_prefix: Option>, /// Represents a connection to a SV2 Upstream role. pub(super) connection: UpstreamConnection, /// Receives SV2 `SubmitSharesExtended` messages translated from SV1 `mining.submit` messages. @@ -108,38 +79,12 @@ pub struct Upstream { /// Sends SV2 `NewExtendedMiningJob` messages to be translated (along with SV2 `SetNewPrevHash` /// messages) into SV1 `mining.notify` messages. Received and translated by the `Bridge`. tx_sv2_new_ext_mining_job: Sender>, - /// Sends the extranonce1 and the channel id received in the SV2 - /// `OpenExtendedMiningChannelSuccess` message to be used by the `Downstream` and sent to - /// the Downstream role in a SV2 `mining.subscribe` response message. Passed to the - /// `Downstream` on connection creation. - tx_sv2_extranonce: Sender<(ExtendedExtranonce, u32)>, /// This allows the upstream threads to be able to communicate back to the main thread its /// current status. tx_status: status::Sender, - /// The first `target` is received by the Upstream role in the SV2 - /// `OpenExtendedMiningChannelSuccess` message, then updated periodically via SV2 `SetTarget` - /// messages. Passed to the `Downstream` on connection creation and sent to the Downstream role - /// via the SV1 `mining.set_difficulty` message. - target: Arc>>, - /// Tracks the most recently sent nominal hashrate to prevent unnecessary updates. - pub last_sent_hashrate: Option, - /// Minimum `extranonce2` size. Initially requested in the `proxy-config.toml`, and ultimately - /// set by the SV2 Upstream via the SV2 `OpenExtendedMiningChannelSuccess` message. - pub min_extranonce_size: u16, - /// The size of the extranonce1 provided by the upstream role. - pub upstream_extranonce1_size: usize, - // values used to update the channel with the correct nominal hashrate. - // each Downstream instance will add and subtract their hashrates as needed - // and the upstream just needs to occasionally check if it has changed more than - // than the configured percentage - pub(super) difficulty_config: Arc>, task_collector: Arc>>, -} - -impl PartialEq for Upstream { - fn eq(&self, other: &Self) -> bool { - self.channel_id == other.channel_id - } + pub(super) upstream_channel_manager: Arc>, + rx_open_upstream_channel: Receiver, } impl Upstream { @@ -155,12 +100,12 @@ impl Upstream { rx_sv2_submit_shares_ext: Receiver>, tx_sv2_set_new_prev_hash: Sender>, tx_sv2_new_ext_mining_job: Sender>, - min_extranonce_size: u16, - tx_sv2_extranonce: Sender<(ExtendedExtranonce, u32)>, tx_status: status::Sender, - target: Arc>>, - difficulty_config: Arc>, task_collector: Arc>>, + upstream_channel_manager: Arc>, + min_version: u16, + max_version: u16, + rx_open_upstream_channel: Receiver, ) -> ProxyResult<'static, Arc>> { // Connect to the SV2 Upstream role retry connection every 5 seconds. let socket = loop { @@ -191,44 +136,10 @@ impl Upstream { .unwrap(); // Initialize `UpstreamConnection` with channel for SV2 Upstream role communication and // channel for downstream Translator Proxy communication - let connection = UpstreamConnection { receiver, sender }; + let mut connection = UpstreamConnection { receiver, sender }; - Ok(Arc::new(Mutex::new(Self { - connection, - rx_sv2_submit_shares_ext, - extranonce_prefix: None, - tx_sv2_set_new_prev_hash, - tx_sv2_new_ext_mining_job, - channel_id: None, - job_id: None, - last_job_id: None, - min_extranonce_size, - upstream_extranonce1_size: 16, /* 16 is the default since that is the only value the - * pool supports currently */ - tx_sv2_extranonce, - tx_status, - target, - last_sent_hashrate: None, - difficulty_config, - task_collector, - }))) - } - - /// Performs the SV2 connection setup handshake with the Upstream role. - /// - /// Sends a `SetupConnection` message specifying supported protocol versions - /// and flags. Waits for the upstream to respond with either `SetupConnectionSuccess` - /// or `SetupConnectionError`.Upon successful setup, it then sends an - /// `OpenExtendedMiningChannel` request to establish a mining channel, including the - /// negotiated minimum extranonce size and initial nominal hashrate. - pub async fn connect( - self_: Arc>, - min_version: u16, - max_version: u16, - ) -> ProxyResult<'static, ()> { // Get the `SetupConnection` message with Mining Device information (currently hard coded) let setup_connection = Self::get_setup_connection_message(min_version, max_version, false)?; - let mut connection = self_.safe_lock(|s| s.connection.clone())?; // Put the `SetupConnection` message in a `StdFrame` to be sent over the wire let sv2_frame: StdFrame = Message::Common(setup_connection.into()).try_into()?; @@ -257,42 +168,80 @@ impl Upstream { // Gets the message payload let payload = incoming.payload(); + let upstream = Arc::new(Mutex::new(Self { + connection, + rx_sv2_submit_shares_ext, + tx_sv2_set_new_prev_hash, + tx_sv2_new_ext_mining_job, + tx_status, + task_collector, + upstream_channel_manager, + rx_open_upstream_channel, + })); + // Handle the incoming message (should be either `SetupConnectionSuccess` or // `SetupConnectionError`) ParseCommonMessagesFromUpstream::handle_message_common( - self_.clone(), + upstream.clone(), message_type, payload, )?; - // Send open channel request before returning - let nominal_hash_rate = self_.safe_lock(|u| { - u.difficulty_config - .safe_lock(|c| c.channel_nominal_hashrate) - .map_err(|_e| PoisonLock) - })??; - let user_identity = "ABC".to_string().try_into()?; - - // Get the min_extranonce_size from the instance - let min_extranonce_size = self_.safe_lock(|u| u.min_extranonce_size)?; - - let open_channel = Mining::OpenExtendedMiningChannel(OpenExtendedMiningChannel { - request_id: 0, // TODO - user_identity, // TODO - nominal_hash_rate, - max_target: u256_from_int(u64::MAX), // TODO - min_extranonce_size, - }); + Ok(upstream) + } - // reset channel hashrate so downstreams can manage from now on out - self_.safe_lock(|u| { - u.difficulty_config - .safe_lock(|d| d.channel_nominal_hashrate = 0.0) - .map_err(|_e| PoisonLock) - })??; + /// Performs the SV2 connection setup handshake with the Upstream role. + /// + /// Sends a `SetupConnection` message specifying supported protocol versions + /// and flags. Waits for the upstream to respond with either `SetupConnectionSuccess` + /// or `SetupConnectionError`.Upon successful setup, it then sends an + /// `OpenExtendedMiningChannel` request to establish a mining channel, including the + /// negotiated minimum extranonce size and initial nominal hashrate. + pub async fn connect(self_: Arc>) -> ProxyResult<'static, ()> { + let (mut connection, rx_open_upstream_channel) = + self_.safe_lock(|u| (u.connection.clone(), u.rx_open_upstream_channel.clone()))?; + info!("Starting the upstream connection thread"); + tokio::spawn(async move { + loop { + match rx_open_upstream_channel.recv().await { + Ok(open) => { + info!("Received new connection request: {:?}", open); + // Send open channel request before returning + let (nominal_hash_rate, min_extranonce_size) = self_ + .safe_lock(|u| { + u.upstream_channel_manager + .safe_lock(|u| { + (u.bootstrap_nominal_hashrate, u.min_extranonce_size) + }) + .map_err(|_e| PoisonLock) + }) + .unwrap() + .unwrap(); + let user_identity = open.user_identity.try_into().unwrap(); + + let open_channel: Mining<'_> = + Mining::OpenExtendedMiningChannel(OpenExtendedMiningChannel { + request_id: open.request_id, + user_identity, + nominal_hash_rate, + max_target: u256_from_int(u64::MAX), + min_extranonce_size, + }); + + info!( + "Sending open channel message to upstream: {:?}", + open_channel + ); - let sv2_frame: StdFrame = Message::Mining(open_channel).try_into()?; - connection.send(sv2_frame).await?; + let sv2_frame: StdFrame = Message::Mining(open_channel).try_into().unwrap(); + connection.send(sv2_frame).await.unwrap(); + } + Err(e) => { + warn!("Received and error while sending receiving open channel request from bridge: {:?}", e); + } + } + } + }); Ok(()) } @@ -311,23 +260,16 @@ impl Upstream { let task_collector = self_.safe_lock(|s| s.task_collector.clone()).unwrap(); let collector1 = task_collector.clone(); let collector2 = task_collector.clone(); - let ( - tx_frame, - tx_sv2_extranonce, - tx_sv2_new_ext_mining_job, - tx_sv2_set_new_prev_hash, - recv, - tx_status, - ) = clone.safe_lock(|s| { - ( - s.connection.sender.clone(), - s.tx_sv2_extranonce.clone(), - s.tx_sv2_new_ext_mining_job.clone(), - s.tx_sv2_set_new_prev_hash.clone(), - s.connection.receiver.clone(), - s.tx_status.clone(), - ) - })?; + let (tx_frame, tx_sv2_new_ext_mining_job, tx_sv2_set_new_prev_hash, recv, tx_status) = + clone.safe_lock(|s| { + ( + s.connection.sender.clone(), + s.tx_sv2_new_ext_mining_job.clone(), + s.tx_sv2_set_new_prev_hash.clone(), + s.connection.receiver.clone(), + s.tx_status.clone(), + ) + })?; { let self_ = self_.clone(); let tx_status = tx_status.clone(); @@ -385,49 +327,10 @@ impl Upstream { // Does not send the messages anywhere, but instead handle them internally Ok(SendTo::None(Some(m))) => { match m { - Mining::OpenExtendedMiningChannelSuccess(m) => { - let prefix_len = m.extranonce_prefix.len(); - // update upstream_extranonce1_size for tracking - let miner_extranonce2_size = self_ - .safe_lock(|u| { - u.upstream_extranonce1_size = prefix_len; - u.min_extranonce_size as usize - }) - .map_err(|_e| PoisonLock); - let miner_extranonce2_size = - handle_result!(tx_status, miner_extranonce2_size); - let extranonce_prefix: Extranonce = m.extranonce_prefix.into(); - // Create the extended extranonce that will be saved in bridge and - // it will be used to open downstream (sv1) channels - // range 0 is the extranonce1 from upstream - // range 1 is the extranonce1 added by the tproxy - // range 2 is the extranonce2 used by the miner for rolling - // range 0 + range 1 is the extranonce1 sent to the miner - let tproxy_e1_len = super::super::utils::proxy_extranonce1_len( - m.extranonce_size as usize, - miner_extranonce2_size, - ); - let range_0 = 0..prefix_len; // upstream extranonce1 - let range_1 = prefix_len..prefix_len + tproxy_e1_len; // downstream extranonce1 - let range_2 = prefix_len + tproxy_e1_len - ..prefix_len + m.extranonce_size as usize; // extranonce2 - let extended = handle_result!(tx_status, ExtendedExtranonce::from_upstream_extranonce( - extranonce_prefix.clone(), range_0.clone(), range_1.clone(), range_2.clone(), - ).map_err(|err| InvalidExtranonce(format!("Impossible to create a valid extended extranonce from {:?} {:?} {:?} {:?}: {:?}", - extranonce_prefix, range_0, range_1, range_2, err)))); - handle_result!( - tx_status, - tx_sv2_extranonce.send((extended, m.channel_id)).await - ); + Mining::OpenExtendedMiningChannelSuccess(_m) => { + info!("Open extended mining channel success received"); } Mining::NewExtendedMiningJob(m) => { - let job_id = m.job_id; - let res = self_ - .safe_lock(|s| { - let _ = s.job_id.insert(job_id); - }) - .map_err(|_e| PoisonLock); - handle_result!(tx_status, res); handle_result!(tx_status, tx_sv2_new_ext_mining_job.send(m).await); } Mining::SetNewPrevHash(m) => { @@ -478,33 +381,6 @@ impl Upstream { Ok(()) } - // Retrieves the current job ID. - // - // If work selection is enabled (which it is not for a Translator Proxy), - // it would return the last `SetCustomMiningJobSuccess` job ID. If - // work selection is disabled, it returns the job ID from the last - // `NewExtendedMiningJob` - #[allow(clippy::result_large_err)] - fn get_job_id( - self_: &Arc>, - ) -> Result>, super::super::error::Error<'static>> - { - self_ - .safe_lock(|s| { - if s.is_work_selection_enabled() { - s.last_job_id - .ok_or(super::super::error::Error::RolesSv2Logic( - RolesLogicError::NoValidTranslatorJob, - )) - } else { - s.job_id.ok_or(super::super::error::Error::RolesSv2Logic( - RolesLogicError::NoValidJob, - )) - } - }) - .map_err(|_e| PoisonLock) - } - /// Spawns a task to handle outgoing `SubmitSharesExtended` messages. /// /// This task continuously receives `SubmitSharesExtended` messages from the @@ -526,22 +402,9 @@ impl Upstream { let handle_submit = tokio::task::spawn(async move { loop { - let mut sv2_submit: SubmitSharesExtended = + let sv2_submit: SubmitSharesExtended = handle_result!(tx_status, receiver.recv().await); - let channel_id = self_ - .safe_lock(|s| { - s.channel_id - .ok_or(super::super::error::Error::RolesSv2Logic( - RolesLogicError::NotFoundChannelId, - )) - }) - .map_err(|_e| PoisonLock); - sv2_submit.channel_id = - handle_result!(tx_status, handle_result!(tx_status, channel_id)); - let job_id = Self::get_job_id(&self_); - sv2_submit.job_id = handle_result!(tx_status, handle_result!(tx_status, job_id)); - let message = Message::Mining( roles_logic_sv2::parsers::Mining::SubmitSharesExtended(sv2_submit), ); @@ -558,362 +421,4 @@ impl Upstream { Ok(()) } - - // Unimplemented method to check if a submitted share is contained within the upstream target. - // - // This method is currently unimplemented (`todo!()`). Its purpose would be - // to validate a share against the target set by the upstream pool. - fn _is_contained_in_upstream_target(&self, _share: SubmitSharesExtended) -> bool { - todo!() - } - - // Creates the initial `SetupConnection` message for the SV2 handshake. - // - // This message contains information about the proxy acting as a mining device, - // including supported protocol versions, flags, and hardcoded endpoint details. - // - // TODO: The Mining Device information is currently hardcoded. It should ideally - // be configurable or derived from the downstream connections. - #[allow(clippy::result_large_err)] - fn get_setup_connection_message( - min_version: u16, - max_version: u16, - is_work_selection_enabled: bool, - ) -> ProxyResult<'static, SetupConnection<'static>> { - let endpoint_host = "0.0.0.0".to_string().into_bytes().try_into()?; - let vendor = String::new().try_into()?; - let hardware_version = String::new().try_into()?; - let firmware = String::new().try_into()?; - let device_id = String::new().try_into()?; - let flags = match is_work_selection_enabled { - false => 0b0000_0000_0000_0000_0000_0000_0000_0100, - true => 0b0000_0000_0000_0000_0000_0000_0000_0110, - }; - Ok(SetupConnection { - protocol: Protocol::MiningProtocol, - min_version, - max_version, - flags, - endpoint_host, - endpoint_port: 50, - vendor, - hardware_version, - firmware, - device_id, - }) - } -} - -// Can be removed? -impl IsUpstream for Upstream { - fn get_version(&self) -> u16 { - todo!() - } - - fn get_flags(&self) -> u32 { - todo!() - } - - fn get_supported_protocols(&self) -> Vec { - todo!() - } - - fn get_id(&self) -> u32 { - todo!() - } - - fn get_mapper(&mut self) -> Option<&mut roles_logic_sv2::common_properties::RequestIdMapper> { - todo!() - } -} - -// Can be removed? -impl IsMiningUpstream for Upstream { - fn total_hash_rate(&self) -> u64 { - todo!() - } - - fn add_hash_rate(&mut self, _to_add: u64) { - todo!() - } - - fn get_opened_channels( - &mut self, - ) -> &mut Vec { - todo!() - } - - fn update_channels(&mut self, _c: roles_logic_sv2::common_properties::UpstreamChannel) { - todo!() - } -} - -impl ParseCommonMessagesFromUpstream for Upstream { - // Handles the SV2 `SetupConnectionSuccess` message received from the upstream. - // - // Returns `Ok(SendToCommon::None(None))` as this message is handled internally - // and does not require a direct response or forwarding. - fn handle_setup_connection_success( - &mut self, - m: roles_logic_sv2::common_messages_sv2::SetupConnectionSuccess, - ) -> Result { - info!( - "Received `SetupConnectionSuccess`: version={}, flags={:b}", - m.used_version, m.flags - ); - Ok(SendToCommon::None(None)) - } - - fn handle_setup_connection_error( - &mut self, - _: roles_logic_sv2::common_messages_sv2::SetupConnectionError, - ) -> Result { - todo!() - } - - fn handle_channel_endpoint_changed( - &mut self, - _: roles_logic_sv2::common_messages_sv2::ChannelEndpointChanged, - ) -> Result { - todo!() - } - - fn handle_reconnect(&mut self, _m: Reconnect) -> Result { - todo!() - } -} - -/// Connection-wide SV2 Upstream role messages parser implemented by a downstream ("downstream" -/// here is relative to the SV2 Upstream role and is represented by this `Upstream` struct). -impl ParseMiningMessagesFromUpstream for Upstream { - /// Returns the type of channel used between this proxy and the SV2 Upstream. - /// For a Translator Proxy, this is always `Extended`. - fn get_channel_type(&self) -> SupportedChannelTypes { - SupportedChannelTypes::Extended - } - - /// Indicates whether work selection is enabled for this upstream connection. - /// For a Translator Proxy, work selection is handled by the upstream pool, - /// so this method always returns `false`. - fn is_work_selection_enabled(&self) -> bool { - false - } - - /// The SV2 `OpenStandardMiningChannelSuccess` message is NOT handled because it is NOT used - /// for the Translator Proxy as only `Extended` channels are used between the SV1/SV2 Translator - /// Proxy and the SV2 Upstream role. - fn handle_open_standard_mining_channel_success( - &mut self, - _m: roles_logic_sv2::mining_sv2::OpenStandardMiningChannelSuccess, - ) -> Result, RolesLogicError> { - panic!("Standard Mining Channels are not used in Translator Proxy") - } - - /// Handles the SV2 `OpenExtendedMiningChannelSuccess` message. - /// - /// This message is received after requesting to open an extended mining channel. - /// It provides the assigned `channel_id`, the extranonce prefix, the initial - /// mining `target`, and the expected `extranonce_size`. It stores the `channel_id` and - /// `extranonce_prefix`, updates the shared `target`, and prepares the extranonce - /// information (including calculating the size for the TProxy's added extranonce1) to be - /// sent to the Downstream handler for use with SV1 clients. - /// - /// Returns `Ok(SendTo::None(Some(Mining::OpenExtendedMiningChannelSuccess)))` - /// to indicate that the message has been handled internally and should be - /// forwarded to the Bridge. - fn handle_open_extended_mining_channel_success( - &mut self, - m: roles_logic_sv2::mining_sv2::OpenExtendedMiningChannelSuccess, - ) -> Result, RolesLogicError> { - info!( - "Received OpenExtendedMiningChannelSuccess with request id: {} and channel id: {}", - m.request_id, m.channel_id - ); - debug!("OpenStandardMiningChannelSuccess: {:?}", m); - let tproxy_e1_len = super::super::utils::proxy_extranonce1_len( - m.extranonce_size as usize, - self.min_extranonce_size.into(), - ) as u16; - if self.min_extranonce_size + tproxy_e1_len < m.extranonce_size { - return Err(RolesLogicError::InvalidExtranonceSize( - self.min_extranonce_size, - m.extranonce_size, - )); - } - self.target.safe_lock(|t| *t = m.target.to_vec())?; - - info!("Up: Successfully Opened Extended Mining Channel"); - self.channel_id = Some(m.channel_id); - self.extranonce_prefix = Some(m.extranonce_prefix.to_vec()); - let m = Mining::OpenExtendedMiningChannelSuccess(m.into_static()); - Ok(SendTo::None(Some(m))) - } - - /// Handles the SV2 `OpenExtendedMiningChannelError` message (TODO). - fn handle_open_mining_channel_error( - &mut self, - m: roles_logic_sv2::mining_sv2::OpenMiningChannelError, - ) -> Result, RolesLogicError> { - error!( - "Received OpenExtendedMiningChannelError with error code {}", - std::str::from_utf8(m.error_code.as_ref()).unwrap_or("unknown error code") - ); - Ok(SendTo::None(Some(Mining::OpenMiningChannelError( - m.as_static(), - )))) - } - - /// Handles the SV2 `UpdateChannelError` message (TODO). - fn handle_update_channel_error( - &mut self, - m: roles_logic_sv2::mining_sv2::UpdateChannelError, - ) -> Result, RolesLogicError> { - error!( - "Received UpdateChannelError with error code {}", - std::str::from_utf8(m.error_code.as_ref()).unwrap_or("unknown error code") - ); - Ok(SendTo::None(Some(Mining::UpdateChannelError( - m.as_static(), - )))) - } - - /// Handles the SV2 `CloseChannel` message (TODO). - fn handle_close_channel( - &mut self, - m: roles_logic_sv2::mining_sv2::CloseChannel, - ) -> Result, RolesLogicError> { - info!("Received CloseChannel for channel id: {}", m.channel_id); - Ok(SendTo::None(Some(Mining::CloseChannel(m.as_static())))) - } - - /// Handles the SV2 `SetExtranoncePrefix` message (TODO). - fn handle_set_extranonce_prefix( - &mut self, - _: roles_logic_sv2::mining_sv2::SetExtranoncePrefix, - ) -> Result, RolesLogicError> { - todo!() - } - - /// Handles the SV2 `SubmitSharesSuccess` message. - fn handle_submit_shares_success( - &mut self, - m: roles_logic_sv2::mining_sv2::SubmitSharesSuccess, - ) -> Result, RolesLogicError> { - info!("Received SubmitSharesSuccess"); - debug!("SubmitSharesSuccess: {:?}", m); - Ok(SendTo::None(None)) - } - - /// Handles the SV2 `SubmitSharesError` message. - fn handle_submit_shares_error( - &mut self, - m: roles_logic_sv2::mining_sv2::SubmitSharesError, - ) -> Result, RolesLogicError> { - error!( - "Received SubmitSharesError with error code {}", - std::str::from_utf8(m.error_code.as_ref()).unwrap_or("unknown error code") - ); - Ok(SendTo::None(None)) - } - - /// The SV2 `NewMiningJob` message is NOT handled because it is NOT used for the Translator - /// Proxy as only `Extended` channels are used between the SV1/SV2 Translator Proxy and the SV2 - /// Upstream role. - fn handle_new_mining_job( - &mut self, - _m: roles_logic_sv2::mining_sv2::NewMiningJob, - ) -> Result, RolesLogicError> { - panic!("Standard Mining Channels are not used in Translator Proxy") - } - - /// Handles the SV2 `NewExtendedMiningJob` message which is used (along with the SV2 - /// `SetNewPrevHash` message) to later create a SV1 `mining.notify` for the Downstream - /// role. - fn handle_new_extended_mining_job( - &mut self, - m: NewExtendedMiningJob, - ) -> Result, RolesLogicError> { - info!( - "Received new extended mining job for channel id: {} with job id: {} is_future: {}", - m.channel_id, - m.job_id, - m.is_future() - ); - debug!("NewExtendedMiningJob: {:?}", m); - if self.is_work_selection_enabled() { - Ok(SendTo::None(None)) - } else { - IS_NEW_JOB_HANDLED.store(false, std::sync::atomic::Ordering::SeqCst); - if !m.version_rolling_allowed { - warn!("VERSION ROLLING NOT ALLOWED IS A TODO"); - // todo!() - } - - let message = Mining::NewExtendedMiningJob(m.into_static()); - - Ok(SendTo::None(Some(message))) - } - } - - /// Handles the SV2 `SetNewPrevHash` message which is used (along with the SV2 - /// `NewExtendedMiningJob` message) to later create a SV1 `mining.notify` for the Downstream - /// role. - fn handle_set_new_prev_hash( - &mut self, - m: SetNewPrevHash, - ) -> Result, RolesLogicError> { - info!( - "Received SetNewPrevHash channel id: {}, job id: {}", - m.channel_id, m.job_id - ); - debug!("SetNewPrevHash: {:?}", m); - if self.is_work_selection_enabled() { - Ok(SendTo::None(None)) - } else { - let message = Mining::SetNewPrevHash(m.into_static()); - Ok(SendTo::None(Some(message))) - } - } - - /// Handles the SV2 `SetCustomMiningJobSuccess` message (TODO). - fn handle_set_custom_mining_job_success( - &mut self, - m: roles_logic_sv2::mining_sv2::SetCustomMiningJobSuccess, - ) -> Result, RolesLogicError> { - info!( - "Received SetCustomMiningJobSuccess for channel id: {} for job id: {}", - m.channel_id, m.job_id - ); - debug!("SetCustomMiningJobSuccess: {:?}", m); - self.last_job_id = Some(m.job_id); - Ok(SendTo::None(None)) - } - - /// Handles the SV2 `SetCustomMiningJobError` message (TODO). - fn handle_set_custom_mining_job_error( - &mut self, - _m: roles_logic_sv2::mining_sv2::SetCustomMiningJobError, - ) -> Result, RolesLogicError> { - unimplemented!() - } - - /// Handles the SV2 `SetTarget` message which updates the Downstream role(s) target - /// difficulty via the SV1 `mining.set_difficulty` message. - fn handle_set_target( - &mut self, - m: roles_logic_sv2::mining_sv2::SetTarget, - ) -> Result, RolesLogicError> { - info!("Received SetTarget for channel id: {}", m.channel_id); - debug!("SetTarget: {:?}", m); - let m = m.into_static(); - self.target.safe_lock(|t| *t = m.maximum_target.to_vec())?; - Ok(SendTo::None(None)) - } - - fn handle_set_group_channel( - &mut self, - _m: SetGroupChannel, - ) -> Result, RolesLogicError> { - todo!() - } }