From 211affb5f5ba34a5a8e1c7e267b0dcc27a283c3f Mon Sep 17 00:00:00 2001 From: bit-aloo Date: Tue, 21 Jul 2026 17:43:55 +0530 Subject: [PATCH 1/3] migrate translator to new shared primitives API --- miner-apps/translator/src/lib/monitoring.rs | 102 +-- .../translator/src/lib/sv1/downstream.rs | 79 ++- .../translator/src/lib/sv1/monitoring.rs | 67 +- .../lib/sv1/sv1_server/difficulty_manager.rs | 480 ++++++++------- .../sv1_server/downstream_message_handler.rs | 309 ++++++---- .../translator/src/lib/sv1/sv1_server/mod.rs | 581 ++++++++++-------- .../extensions_message_handler.rs | 12 +- .../channel_manager/mining_message_handler.rs | 552 +++++++++-------- .../src/lib/sv2/channel_manager/mod.rs | 166 ++--- 9 files changed, 1280 insertions(+), 1068 deletions(-) diff --git a/miner-apps/translator/src/lib/monitoring.rs b/miner-apps/translator/src/lib/monitoring.rs index c7e6ae28f..eeb02c453 100644 --- a/miner-apps/translator/src/lib/monitoring.rs +++ b/miner-apps/translator/src/lib/monitoring.rs @@ -18,53 +18,53 @@ impl ServerMonitoring for ChannelManager { TproxyMode::Aggregated => { // In Aggregated mode: one shared channel to the server // stored under AGGREGATED_CHANNEL_ID - if let Some(aggregated_extended_channel) = - self.extended_channels.get(&AGGREGATED_CHANNEL_ID) - { - let channel_id = aggregated_extended_channel.get_channel_id(); - let target = *aggregated_extended_channel.get_target(); - let extranonce_prefix = - aggregated_extended_channel.get_extranonce_prefix().to_vec(); - let user_identity = aggregated_extended_channel.get_user_identity().to_string(); - let full_extranonce_size = - aggregated_extended_channel.get_full_extranonce_size(); - let rollable_extranonce_size = - aggregated_extended_channel.get_rollable_extranonce_size(); - let version_rolling = aggregated_extended_channel.is_version_rolling(); - let nominal_hashrate = aggregated_extended_channel.get_nominal_hashrate(); - let share_accounting = aggregated_extended_channel.get_share_accounting(); - let shares_rejected_by_reason = share_accounting - .get_rejected_shares() - .map(|(reason, count)| (reason.to_string(), count)) - .collect(); - let shares_rejected = share_accounting.get_rejected_shares_count(); - - extended_channels.push(ServerExtendedChannelInfo { - channel_id, - user_identity, - nominal_hashrate: report_hashrate.then_some(nominal_hashrate), - target_hex: hex::encode(target.to_be_bytes()), - extranonce_prefix_hex: hex::encode(extranonce_prefix), - full_extranonce_size, - rollable_extranonce_size, - version_rolling, - shares_acknowledged: share_accounting.get_acknowledged_shares(), - shares_submitted: share_accounting.get_validated_shares(), - shares_rejected, - shares_rejected_by_reason, - acknowledged_work_sum: share_accounting.get_acknowledged_work_sum(), - validated_work_sum: share_accounting.get_validated_work_sum(), - best_diff: share_accounting.get_best_diff(), - blocks_found: share_accounting.get_blocks_found(), - }); - } + self.extended_channels.with( + &AGGREGATED_CHANNEL_ID, + |aggregated_extended_channel| { + let channel_id = aggregated_extended_channel.get_channel_id(); + let target = *aggregated_extended_channel.get_target(); + let extranonce_prefix = + aggregated_extended_channel.get_extranonce_prefix().to_vec(); + let user_identity = + aggregated_extended_channel.get_user_identity().to_string(); + let full_extranonce_size = + aggregated_extended_channel.get_full_extranonce_size(); + let rollable_extranonce_size = + aggregated_extended_channel.get_rollable_extranonce_size(); + let version_rolling = aggregated_extended_channel.is_version_rolling(); + let nominal_hashrate = aggregated_extended_channel.get_nominal_hashrate(); + let share_accounting = aggregated_extended_channel.get_share_accounting(); + let shares_rejected_by_reason = share_accounting + .get_rejected_shares() + .map(|(reason, count)| (reason.to_string(), count)) + .collect(); + let shares_rejected = share_accounting.get_rejected_shares_count(); + + extended_channels.push(ServerExtendedChannelInfo { + channel_id, + user_identity, + nominal_hashrate: report_hashrate.then_some(nominal_hashrate), + target_hex: hex::encode(target.to_be_bytes()), + extranonce_prefix_hex: hex::encode(extranonce_prefix), + full_extranonce_size, + rollable_extranonce_size, + version_rolling, + shares_acknowledged: share_accounting.get_acknowledged_shares(), + shares_submitted: share_accounting.get_validated_shares(), + shares_rejected, + shares_rejected_by_reason, + acknowledged_work_sum: share_accounting.get_acknowledged_work_sum(), + validated_work_sum: share_accounting.get_validated_work_sum(), + best_diff: share_accounting.get_best_diff(), + blocks_found: share_accounting.get_blocks_found(), + }); + }, + ); } TproxyMode::NonAggregated => { // In NonAggregated mode: each downstream Sv1 miner has its own upstream Sv2 // channel to the server - for channel in self.extended_channels.iter() { - let extended_channel = channel.value(); - + self.extended_channels.for_each(|_, extended_channel| { let channel_id = extended_channel.get_channel_id(); let target = extended_channel.get_target(); let extranonce_prefix = extended_channel.get_extranonce_prefix(); @@ -98,7 +98,7 @@ impl ServerMonitoring for ChannelManager { best_diff: share_accounting.get_best_diff(), blocks_found: share_accounting.get_blocks_found(), }); - } + }); } } @@ -163,14 +163,16 @@ mod tests { manager .extended_channels - .get_mut(&AGGREGATED_CHANNEL_ID) - .unwrap() - .on_share_acknowledgement(2, 10); + .with_mut(&AGGREGATED_CHANNEL_ID, |channel| { + channel.on_share_acknowledgement(2, 10); + }) + .unwrap(); manager .extended_channels - .get_mut(&7) - .unwrap() - .on_share_acknowledgement(5, 25); + .with_mut(&7, |channel| { + channel.on_share_acknowledgement(5, 25); + }) + .unwrap(); let server = manager.get_server(); let aggregated = server.extended_channels.first().unwrap(); diff --git a/miner-apps/translator/src/lib/sv1/downstream.rs b/miner-apps/translator/src/lib/sv1/downstream.rs index fe7d2599e..981ffd377 100644 --- a/miner-apps/translator/src/lib/sv1/downstream.rs +++ b/miner-apps/translator/src/lib/sv1/downstream.rs @@ -15,7 +15,6 @@ use std::{ }; use stratum_apps::{ channel_utils::ReceiverCleanup, - custom_mutex::Mutex, fallback_coordinator::FallbackCoordinator, stratum_core::{ bitcoin::Target, @@ -25,6 +24,7 @@ use stratum_apps::{ utils::{Extranonce, HexU32Be}, }, }, + sync::SharedLock, task_manager::TaskManager, utils::types::{ChannelId, DownstreamId, Hashrate}, }; @@ -169,7 +169,7 @@ impl DownstreamData { #[derive(Clone, Debug)] pub struct Downstream { pub downstream_id: DownstreamId, - pub downstream_data: Arc>, + pub downstream_data: SharedLock, pub downstream_io: DownstreamIo, // Flag to track if SV1 handshake is complete (subscribe + authorize) pub sv1_handshake_complete: Arc, @@ -258,12 +258,12 @@ impl Downstream { #[cfg(feature = "monitoring")] connection_ip: IpAddr, downstream_cancellation_token: CancellationToken, ) -> Self { - let downstream_data = Arc::new(Mutex::new(DownstreamData::new( + let downstream_data = SharedLock::new(DownstreamData::new( hashrate, target, #[cfg(feature = "monitoring")] connection_ip, - ))); + )); let downstream_channel_io = DownstreamIo::new( downstream_sv1_sender, downstream_sv1_receiver, @@ -399,14 +399,15 @@ impl Downstream { // Cache the Sv1 set_difficulty message to be sent before the next // notify debug!("Down: Caching mining.set_difficulty to send before next mining.notify"); - self.downstream_data.super_safe_lock(|d| { - d.cached_set_difficulty = Some(message); - }); + self.downstream_data + .with(|d| d.cached_set_difficulty = Some(message)) + .map_err(TproxyError::shutdown)?; return Ok(()); } "mining.notify" => { - let (pending_set_difficulty, notify_opt) = - self.downstream_data.super_safe_lock(|d| { + let (pending_set_difficulty, notify_opt) = self + .downstream_data + .with(|d| { let cached_set_difficulty = d.cached_set_difficulty.take(); // Prepare the notify message and update state @@ -437,7 +438,8 @@ impl Downstream { } else { (cached_set_difficulty, None) } - }); + }) + .map_err(TproxyError::shutdown)?; if let Some(set_difficulty_msg) = &pending_set_difficulty { debug!("Down: Sending pending mining.set_difficulty before mining.notify"); @@ -490,19 +492,22 @@ impl Downstream { match notification.method.as_str() { "mining.set_difficulty" => { debug!("Down: SV1 handshake not complete, caching mining.set_difficulty"); - self.downstream_data.super_safe_lock(|d| { - d.cached_set_difficulty = Some(message); - }); + self.downstream_data + .with(|d| d.cached_set_difficulty = Some(message)) + .map_err(TproxyError::shutdown)?; } "mining.notify" => { debug!("Down: SV1 handshake not complete, caching mining.notify"); - self.downstream_data.super_safe_lock(|d| { - d.cached_notify = Some(message.clone()); - let notify = - server_to_client::Notify::try_from(notification.clone()) - .expect("this must be a mining.notify"); - d.last_job_version_field = Some(notify.version.0); - }); + self.downstream_data + .with(|d| { + d.cached_notify = Some(message.clone()); + let notify = server_to_client::Notify::try_from( + notification.clone(), + ) + .expect("this must be a mining.notify"); + d.last_job_version_field = Some(notify.version.0); + }) + .map_err(TproxyError::shutdown)?; } _ => { debug!( @@ -567,8 +572,9 @@ impl Downstream { pub(super) async fn handle_sv1_handshake_completion( &self, ) -> TproxyResult<(), error::Downstream> { - let (cached_set_difficulty, cached_notify, downstream_id) = - self.downstream_data.super_safe_lock(|d| { + let (cached_set_difficulty, cached_notify, downstream_id) = self + .downstream_data + .with(|d| { self.sv1_handshake_complete .store(true, std::sync::atomic::Ordering::SeqCst); ( @@ -576,7 +582,8 @@ impl Downstream { d.cached_notify.take(), self.downstream_id, ) - }); + }) + .map_err(TproxyError::shutdown)?; debug!("Down: SV1 handshake completed for downstream"); // Send cached messages in correct order: set_difficulty first, then notify @@ -595,14 +602,16 @@ impl Downstream { })?; // Update target and hashrate after sending set_difficulty - self.downstream_data.super_safe_lock(|d| { - if let Some(new_target) = d.pending_target.take() { - d.target = new_target; - } - if let Some(new_hashrate) = d.pending_hashrate.take() { - d.hashrate = Some(new_hashrate); - } - }); + self.downstream_data + .with(|d| { + if let Some(new_target) = d.pending_target.take() { + d.target = new_target; + } + if let Some(new_hashrate) = d.pending_hashrate.take() { + d.hashrate = Some(new_hashrate); + } + }) + .map_err(TproxyError::shutdown)?; } if let Some(notify_msg) = cached_notify { @@ -619,9 +628,11 @@ impl Downstream { TproxyError::disconnect(TproxyErrorKind::ChannelErrorSender, downstream_id) })?; // Update last job received time for keepalive tracking - self.downstream_data.super_safe_lock(|d| { - d.last_job_received_time = Some(Instant::now()); - }); + self.downstream_data + .with(|d| { + d.last_job_received_time = Some(Instant::now()); + }) + .map_err(TproxyError::shutdown)?; } Ok(()) diff --git a/miner-apps/translator/src/lib/sv1/monitoring.rs b/miner-apps/translator/src/lib/sv1/monitoring.rs index db758bff3..6f214e7df 100644 --- a/miner-apps/translator/src/lib/sv1/monitoring.rs +++ b/miner-apps/translator/src/lib/sv1/monitoring.rs @@ -31,7 +31,7 @@ fn downstream_to_sv1_client_info( ) -> Option { downstream .downstream_data - .safe_lock(|dd| Sv1ClientInfo { + .with(|dd| Sv1ClientInfo { client_id: downstream.downstream_id, channel_id: dd.channel_id, connection_ip: dd.connection_ip, @@ -59,34 +59,37 @@ fn downstream_to_sv1_client_info( impl Sv1ClientsMonitoring for Sv1Server { fn get_sv1_clients(&self) -> Vec { + let mut clients = Vec::new(); + self.downstreams.for_each(|downstream_id, downstream| { + let miner_telemetry = self.miner_telemetry_for(downstream_id); + let management_ip = self.miner_telemetry_management_ip_for(downstream_id); + let miner_telemetry_status = self.miner_telemetry_status_for(downstream_id); + if let Some(client) = downstream_to_sv1_client_info( + downstream, + miner_telemetry, + management_ip, + miner_telemetry_status, + ) { + clients.push(client); + } + }); + clients + } + + fn get_sv1_client_by_id(&self, client_id: usize) -> Option { + let miner_telemetry = self.miner_telemetry_for(client_id); + let management_ip = self.miner_telemetry_management_ip_for(client_id); + let miner_telemetry_status = self.miner_telemetry_status_for(client_id); self.downstreams - .iter() - .filter_map(|downstream| { - let miner_telemetry = self.miner_telemetry_for(*downstream.key()); - let management_ip = self.miner_telemetry_management_ip_for(*downstream.key()); - let miner_telemetry_status = self.miner_telemetry_status_for(*downstream.key()); + .with(&client_id, |downstream| { downstream_to_sv1_client_info( - downstream.value(), + downstream, miner_telemetry, management_ip, miner_telemetry_status, ) }) - .collect() - } - - fn get_sv1_client_by_id(&self, client_id: usize) -> Option { - let miner_telemetry = self.miner_telemetry_for(client_id); - let management_ip = self.miner_telemetry_management_ip_for(client_id); - let miner_telemetry_status = self.miner_telemetry_status_for(client_id); - self.downstreams.get(&client_id).and_then(|downstream| { - downstream_to_sv1_client_info( - downstream.value(), - miner_telemetry, - management_ip, - miner_telemetry_status, - ) - }) + .flatten() } } @@ -268,15 +271,15 @@ impl Sv1Server { } fn current_downstream_sv1_usernames(&self) -> Vec<(DownstreamId, String)> { - self.downstreams - .iter() - .filter_map(|downstream| { - downstream - .value() - .downstream_data - .safe_lock(|data| (*downstream.key(), data.sv1_username.clone())) - .ok() - }) - .collect() + let mut worker_names = Vec::new(); + self.downstreams.for_each(|downstream_id, downstream| { + if let Ok(worker_name) = downstream + .downstream_data + .with(|data| data.sv1_username.clone()) + { + worker_names.push((downstream_id, worker_name)); + } + }); + worker_names } } diff --git a/miner-apps/translator/src/lib/sv1/sv1_server/difficulty_manager.rs b/miner-apps/translator/src/lib/sv1/sv1_server/difficulty_manager.rs index 525b6c136..cc15bbcdf 100644 --- a/miner-apps/translator/src/lib/sv1/sv1_server/difficulty_manager.rs +++ b/miner-apps/translator/src/lib/sv1/sv1_server/difficulty_manager.rs @@ -1,7 +1,11 @@ use std::sync::Arc; -use crate::sv1::sv1_server::{ - PendingTargetUpdate, SV1_MIN_DIFFICULTY_FOR_INTEGER_POWER_OF_TWO_ROUNDING, +use crate::{ + error::{self, TproxyError, TproxyErrorKind, TproxyResult}, + sv1::{ + sv1_server::{PendingTargetUpdate, SV1_MIN_DIFFICULTY_FOR_INTEGER_POWER_OF_TWO_ROUNDING}, + Sv1Server, + }, }; use stratum_apps::{ @@ -19,8 +23,6 @@ use stratum_apps::{ }; use tracing::{debug, error, info, trace, warn}; -use crate::sv1::Sv1Server; - enum AggregatedSnapshot { Active { total_hashrate: Hashrate, @@ -35,7 +37,7 @@ impl Sv1Server { /// /// This method implements the SV1 server's variable difficulty logic for all downstreams. /// Every 60 seconds, this method updates the difficulty state for each downstream. - pub(super) async fn spawn_vardiff_loop(self: Arc) { + pub(super) async fn spawn_vardiff_loop(self: Arc) -> TproxyResult<(), error::Sv1Server> { info!("Variable difficulty adjustment enabled - starting vardiff loop"); let mut ticker = tokio::time::interval(std::time::Duration::from_secs(60)); @@ -43,7 +45,7 @@ impl Sv1Server { ticker.tick().await; info!("Starting vardiff loop for downstreams"); - self.handle_vardiff_updates().await; + self.handle_vardiff_updates().await?; } } @@ -57,148 +59,186 @@ impl Sv1Server { /// - If new_target < upstream_target: wait for SetTarget response before sending /// set_difficulty /// 4. Handle aggregated vs non-aggregated modes for UpdateChannel messages - async fn handle_vardiff_updates(&self) { + async fn handle_vardiff_updates(&self) -> TproxyResult<(), error::Sv1Server> { let mut immediate_updates = Vec::new(); let mut all_updates = Vec::new(); // All updates will generate UpdateChannel messages - for vardiff_key_pair in self.vardiff.iter() { - let downstream_id = vardiff_key_pair.key(); - let vardiff = vardiff_key_pair.value(); + self.vardiff.try_for_each_mut(|downstream_id, vardiff_state| { debug!("Updating vardiff for downstream_id: {}", downstream_id); - let Some(downstream) = self.downstreams.get(downstream_id) else { - continue; + let (channel_id, hashrate, target, upstream_target) = match self + .with_registered_downstream(downstream_id, |downstream| { + downstream + .downstream_data + .with(|data| { + // It's safe to unwrap hashrate because we know that + // the downstream has a hashrate (we are + // doing vardiff) + ( + data.channel_id, + data.hashrate.unwrap(), + data.target, + data.upstream_target, + ) + }) + .map_err(crate::error::TproxyError::shutdown) + }) { + Ok(snapshot) => snapshot, + Err(e) if matches!(e.kind, TproxyErrorKind::DownstreamNotFound(_)) => { + return Ok(()); + } + Err(e) => return Err(e), }; - let (channel_id, hashrate, target, upstream_target) = - downstream.downstream_data.super_safe_lock(|data| { - // It's safe to unwrap hashrate because we know that - // the downstream has a hashrate (we are - // doing vardiff) - ( - data.channel_id, - data.hashrate.unwrap(), - data.target, - data.upstream_target, - ) - }); let Some(channel_id) = channel_id else { error!("Channel id is none for downstream_id: {}", downstream_id); - continue; + return Ok(()); }; - let new_hashrate_opt = vardiff.super_safe_lock(|state| { - state.try_vardiff(hashrate, &target, self.shares_per_minute) - }); - - if let Ok(Some(new_hashrate)) = new_hashrate_opt { - // Calculate new target based on new hashrate - let new_target: Target = - match hash_rate_to_target(new_hashrate as f64, self.shares_per_minute as f64) { - Ok(target) => target, - Err(e) => { - error!( - "Failed to calculate target for hashrate {}: {:?}", - new_hashrate, e - ); - continue; + let new_hashrate_opt = + vardiff_state.try_vardiff(hashrate, &target, self.shares_per_minute); + + match new_hashrate_opt { + Ok(Some(new_hashrate)) => { + // Calculate new target based on new hashrate + let new_target: Target = hash_rate_to_target( + new_hashrate as f64, + self.shares_per_minute as f64, + ) + .map_err(|e| { + TproxyError::shutdown(TproxyErrorKind::General(format!( + "failed to calculate target for hashrate {new_hashrate}: {e:?}" + ))) + })?; + // Always update the downstream's pending target and hashrate + if let Err(e) = self.with_registered_downstream(downstream_id, |downstream| { + downstream + .downstream_data + .with(|data| { + // Store the advertised (pow2 rounded) target so share + // validation matches the difficulty the miner was sent. + data.set_pending_target( + sv1_advertised_target_from_sv2_target( + new_target, + SV1_MIN_DIFFICULTY_FOR_INTEGER_POWER_OF_TWO_ROUNDING, + ) + .unwrap_or(new_target), + downstream.downstream_id, + ); + data.set_pending_hashrate( + Some(new_hashrate), + downstream.downstream_id, + ); + data.stable_hashrate = false; + }) + .map_err(crate::error::TproxyError::shutdown) + }) { + if matches!(e.kind, TproxyErrorKind::DownstreamNotFound(_)) { + return Ok(()); } - }; - // Always update the downstream's pending target and hashrate - if let Some(d) = self.downstreams.get(downstream_id) { - _ = d.downstream_data.safe_lock(|data| { - // Store the advertised (pow2 rounded) target so share - // validation matches the difficulty the miner was sent. - data.set_pending_target( - sv1_advertised_target_from_sv2_target( - new_target, - SV1_MIN_DIFFICULTY_FOR_INTEGER_POWER_OF_TWO_ROUNDING, - ) - .unwrap_or(new_target), - d.downstream_id, - ); - data.set_pending_hashrate(Some(new_hashrate), d.downstream_id); - data.stable_hashrate = false; - }); - } - // All updates will be sent as UpdateChannel messages - all_updates.push((*downstream_id, channel_id, new_target, new_hashrate)); - // Determine if we should send set_difficulty immediately or wait - match upstream_target { - Some(upstream_target) => { - if new_target >= upstream_target { - // Case 1: new_target >= upstream_target, send set_difficulty - // immediately - trace!( - "✅ Target comparison: new_target ({}) >= upstream_target ({}) for downstream {}, will send set_difficulty immediately", - new_target, upstream_target, downstream_id - ); - immediate_updates.push((channel_id, Some(*downstream_id), new_target)); - } else { - // Case 2: new_target < upstream_target, delay set_difficulty until - // SetTarget + return Err(e); + } + // All updates will be sent as UpdateChannel messages + all_updates.push((downstream_id, channel_id, new_target, new_hashrate)); + // Determine if we should send set_difficulty immediately or wait + match upstream_target { + Some(upstream_target) => { + if new_target >= upstream_target { + // Case 1: new_target >= upstream_target, send set_difficulty + // immediately + trace!( + "✅ Target comparison: new_target ({}) >= upstream_target ({}) for downstream {}, will send set_difficulty immediately", + new_target, upstream_target, downstream_id + ); + immediate_updates.push(( + channel_id, + Some(downstream_id), + new_target, + )); + } else { + // Case 2: new_target < upstream_target, delay set_difficulty until + // SetTarget + trace!( + "⏳ Target comparison: new_target ({}) < upstream_target ({}) for downstream {}, will delay set_difficulty until SetTarget", + new_target, upstream_target, downstream_id + ); + self.pending_target_updates + .with(|data| { + data.push(PendingTargetUpdate { + downstream_id, + new_target, + }) + }) + .map_err(TproxyError::shutdown)?; + } + } + None => { + // No upstream target set yet, send set_difficulty immediately as fallback trace!( - "⏳ Target comparison: new_target ({}) < upstream_target ({}) for downstream {}, will delay set_difficulty until SetTarget", - new_target, upstream_target, downstream_id + "No upstream target set for downstream {}, will send set_difficulty immediately", + downstream_id ); - self.pending_target_updates.super_safe_lock(|data| { - data.push(PendingTargetUpdate { - downstream_id: *downstream_id, - new_target, - }) - }); + immediate_updates.push((channel_id, Some(downstream_id), new_target)); } } - None => { - // No upstream target set yet, send set_difficulty immediately as fallback - trace!( - "No upstream target set for downstream {}, will send set_difficulty immediately", - downstream_id - ); - immediate_updates.push((channel_id, Some(*downstream_id), new_target)); + } + Ok(None) => { + if let Err(e) = self.with_registered_downstream(downstream_id, |downstream| { + downstream + .downstream_data + .with(|data| { + data.stable_hashrate = true; + }) + .map_err(crate::error::TproxyError::shutdown) + }) { + if matches!(e.kind, TproxyErrorKind::DownstreamNotFound(_)) { + return Ok(()); + } + return Err(e); } } - } else if let Ok(None) = new_hashrate_opt { - if let Some(d) = self.downstreams.get(downstream_id) { - _ = d.downstream_data.safe_lock(|data| { - data.stable_hashrate = true; - }); + Err(e) => { + return Err(TproxyError::shutdown(TproxyErrorKind::General(format!( + "failed to update vardiff for downstream {downstream_id}: {e:?}" + )))); } } - } + Ok(()) + })?; // Send UpdateChannel messages for ALL updates (both immediate and delayed) if !all_updates.is_empty() { - self.send_update_channel_messages(all_updates).await; + self.send_update_channel_messages(all_updates).await?; } // Process immediate set_difficulty updates (for new_target >= upstream_target) for (_channel_id, downstream_id, target) in immediate_updates { // Send set_difficulty message immediately - if let Ok(set_difficulty_msg) = + let set_difficulty_msg = build_sv1_set_difficulty_from_sv2_target_with_integer_power_of_two_rounding( target, SV1_MIN_DIFFICULTY_FOR_INTEGER_POWER_OF_TWO_ROUNDING, ) + .map_err(TproxyError::shutdown)?; + let downstream_id = downstream_id.unwrap_or(0); + if let Some(sender) = self + .sv1_server_io + .sv1_server_to_downstream_sender + .with(&downstream_id, |sender| sender.clone()) { - let downstream_id = downstream_id.unwrap_or(0); - if let Some(sender) = self - .sv1_server_io - .sv1_server_to_downstream_sender - .super_safe_lock(|downstream| downstream.get(&downstream_id).cloned()) - { - if let Err(e) = sender.send(set_difficulty_msg).await { - error!( - "Failed to send immediate SetDifficulty message to downstream {}: {:?}", - downstream_id, e - ); - } else { - trace!( - "Sent immediate SetDifficulty to downstream {} (new_target >= upstream_target)", - downstream_id - ); - } - } + sender.send(set_difficulty_msg).await.map_err(|e| { + error!( + "Failed to send immediate SetDifficulty message to downstream {}: {:?}", + downstream_id, e + ); + TproxyError::shutdown(TproxyErrorKind::ChannelErrorSender) + })?; + trace!( + "Sent immediate SetDifficulty to downstream {} (new_target >= upstream_target)", + downstream_id + ); } } + + Ok(()) } /// Sends UpdateChannel messages for all target updates. @@ -213,54 +253,47 @@ impl Sv1Server { * channel_id, * new_target, * new_hashrate) */ - ) { + ) -> TproxyResult<(), error::Sv1Server> { if self.mode.is_aggregated() { // Aggregated mode: Send single UpdateChannel with minimum target and total hashrate of // ALL downstreams - self.send_aggregated_update_channel(all_updates).await; + self.send_aggregated_update_channel(all_updates).await } else { // Non-aggregated mode: Send individual UpdateChannel for each downstream - self.send_non_aggregated_update_channels(all_updates).await; + self.send_non_aggregated_update_channels(all_updates).await } } async fn send_aggregated_update_channel( &self, all_updates: Vec<(DownstreamId, ChannelId, Target, Hashrate)>, - ) { + ) -> TproxyResult<(), error::Sv1Server> { // Nothing to do if we received no updates let Some((_, channel_id, _, _)) = all_updates.first() else { - return; + return Ok(()); }; if self.downstreams.is_empty() { - return; + return Ok(()); } let mut min_target: Option = None; let mut total_hashrate: Hashrate = 0.0; let shares_per_minute = self.shares_per_minute as f64; - for downstream in self.downstreams.iter() { - let downstream_id = *downstream.key(); - let downstream = downstream.value(); - let hashrate = downstream.downstream_data.super_safe_lock(|d| { + self.downstreams.try_for_each(|downstream_id, downstream| { + let hashrate = downstream.downstream_data.with(|d| { d.pending_hashrate .unwrap_or_else(|| d.hashrate.expect("vardiff implies hashrate")) - }); + }).map_err(TproxyError::shutdown)?; // UpdateChannel is upstream-facing, so rebuild the exact target from // hashrate instead of reusing the rounded SV1 advertised target. - let target = match hash_rate_to_target(hashrate as f64, shares_per_minute) { - Ok(target) => target, - Err(e) => { - error!( - "Failed to calculate exact target for downstream {} hashrate {}: {:?}", - downstream_id, hashrate, e - ); - continue; - } - }; + let target = hash_rate_to_target(hashrate as f64, shares_per_minute).map_err(|e| { + TproxyError::shutdown(TproxyErrorKind::General(format!( + "failed to calculate exact target for downstream {downstream_id} hashrate {hashrate}: {e:?}" + ))) + })?; min_target = Some(match min_target { Some(current) => current.min(target), @@ -268,11 +301,12 @@ impl Sv1Server { }); total_hashrate += hashrate; - } + Ok(()) + })?; let Some(min_target) = min_target else { warn!("Skipping aggregated UpdateChannel: no exact downstream target is available"); - return; + return Ok(()); }; let downstream_count = self.downstreams.len(); @@ -291,20 +325,20 @@ impl Sv1Server { all_updates.len() ); - if let Err(e) = self - .sv1_server_io + self.sv1_server_io .channel_manager_sender .send((Mining::UpdateChannel(update_channel), None)) .await - { - error!("Failed to send aggregated UpdateChannel: {:?}", e); - } + .map_err(|e| { + error!("Failed to send aggregated UpdateChannel: {:?}", e); + TproxyError::shutdown(TproxyErrorKind::ChannelErrorSender) + }) } async fn send_non_aggregated_update_channels( &self, all_updates: Vec<(DownstreamId, ChannelId, Target, Hashrate)>, - ) { + ) -> TproxyResult<(), error::Sv1Server> { for (downstream_id, channel_id, new_target, new_hashrate) in all_updates { let update_channel = UpdateChannel { channel_id, @@ -317,18 +351,19 @@ impl Sv1Server { downstream_id, channel_id, new_hashrate, new_target ); - if let Err(e) = self - .sv1_server_io + self.sv1_server_io .channel_manager_sender .send((Mining::UpdateChannel(update_channel), None)) .await - { - error!( - "Failed to send UpdateChannel for downstream {}: {:?}", - downstream_id, e - ); - } + .map_err(|e| { + error!( + "Failed to send UpdateChannel for downstream {}: {:?}", + downstream_id, e + ); + TproxyError::shutdown(TproxyErrorKind::ChannelErrorSender) + })?; } + Ok(()) } /// Handles SetTarget messages from the ChannelManager. @@ -336,7 +371,10 @@ impl Sv1Server { /// Aggregated mode: Single SetTarget updates all downstreams and processes all pending updates /// Non-aggregated mode: Each SetTarget updates one specific downstream and processes its /// pending update - pub(super) async fn handle_set_target_message(&self, set_target: SetTarget<'_>) { + pub(super) async fn handle_set_target_message( + &self, + set_target: SetTarget<'_>, + ) -> TproxyResult<(), error::Sv1Server> { let new_upstream_target = Target::from_le_bytes(set_target.maximum_target.to_array()); debug!( "Received SetTarget for channel {}: new_upstream_target = {}", @@ -350,7 +388,7 @@ impl Sv1Server { } self.handle_non_aggregated_set_target(set_target.channel_id, new_upstream_target) - .await; + .await } /// Handles SetTarget in aggregated mode. @@ -359,22 +397,24 @@ impl Sv1Server { &self, new_upstream_target: Target, channel_id: ChannelId, - ) { + ) -> TproxyResult<(), error::Sv1Server> { debug!("Aggregated mode: Updating upstream target for all downstreams"); - for downstream in self.downstreams.iter() { - let downstream = downstream.value(); - downstream.downstream_data.super_safe_lock(|d| { - d.set_upstream_target(new_upstream_target, downstream.downstream_id); - }); - } + self.downstreams.try_for_each(|_, downstream| { + downstream + .downstream_data + .with(|d| { + d.set_upstream_target(new_upstream_target, downstream.downstream_id); + }) + .map_err(TproxyError::shutdown) + })?; // Process ALL pending difficulty updates that can now be sent downstream let applicable_updates = - self.get_pending_difficulty_updates(new_upstream_target, None, channel_id); + self.get_pending_difficulty_updates(new_upstream_target, None, channel_id)?; self.send_pending_set_difficulty_messages_to_downstream(applicable_updates) - .await; + .await } /// Handles SetTarget in non-aggregated mode. @@ -383,7 +423,7 @@ impl Sv1Server { &self, channel_id: ChannelId, new_upstream_target: Target, - ) { + ) -> TproxyResult<(), error::Sv1Server> { debug!( "Non-aggregated mode: Processing SetTarget for channel {}", channel_id @@ -391,20 +431,25 @@ impl Sv1Server { let Some(downstream_id) = self .channel_id_to_downstream_id - .super_safe_lock(|map| map.get(&channel_id).cloned()) + .with(&channel_id, |downstream_id| *downstream_id) else { warn!("No downstream found for channel {}", channel_id); - return; + return Ok(()); }; - { - let Some(downstream) = self.downstreams.get(&downstream_id) else { + if let Err(e) = self.with_registered_downstream(downstream_id, |downstream| { + downstream + .downstream_data + .with(|d| { + d.set_upstream_target(new_upstream_target, downstream_id); + }) + .map_err(crate::error::TproxyError::shutdown) + }) { + if matches!(e.kind, crate::error::TproxyErrorKind::DownstreamNotFound(_)) { warn!("No downstream found for downstream_id {}", downstream_id); - return; - }; - downstream.downstream_data.super_safe_lock(|d| { - d.set_upstream_target(new_upstream_target, downstream_id); - }); + return Ok(()); + } + return Err(e); } trace!("Updated upstream target for downstream {}", downstream_id); @@ -413,10 +458,10 @@ impl Sv1Server { new_upstream_target, Some(downstream_id), channel_id, - ); + )?; self.send_pending_set_difficulty_messages_to_downstream(applicable_updates) - .await; + .await } /// Gets pending updates that can now be applied based on the new upstream target. @@ -427,10 +472,10 @@ impl Sv1Server { new_upstream_target: Target, downstream_id: Option, channel_id: ChannelId, - ) -> Vec { + ) -> TproxyResult, error::Sv1Server> { let mut applicable_updates = Vec::new(); - self.pending_target_updates.super_safe_lock(|data| { + self.pending_target_updates.with(|data| { data.retain(|pending_update| { // Check if we should process this update let should_process = match downstream_id { @@ -455,54 +500,49 @@ impl Sv1Server { false // remove from pending list (don't keep invalid requests) } }); - }); - applicable_updates + }).map_err(TproxyError::shutdown)?; + Ok(applicable_updates) } /// Sends set_difficulty messages for all applicable pending updates. async fn send_pending_set_difficulty_messages_to_downstream( &self, difficulty_updates: Vec, - ) { + ) -> TproxyResult<(), error::Sv1Server> { for update in difficulty_updates { let set_difficulty_msg = - match build_sv1_set_difficulty_from_sv2_target_with_integer_power_of_two_rounding( + build_sv1_set_difficulty_from_sv2_target_with_integer_power_of_two_rounding( update.new_target, SV1_MIN_DIFFICULTY_FOR_INTEGER_POWER_OF_TWO_ROUNDING, - ) { - Ok(msg) => msg, - Err(e) => { - error!( - "Failed to build SetDifficulty for downstream {}: {:?}", - update.downstream_id, e - ); - continue; - } - }; + ) + .map_err(TproxyError::shutdown)?; if let Some(sender) = self .sv1_server_io .sv1_server_to_downstream_sender - .super_safe_lock(|downstream| downstream.get(&update.downstream_id).cloned()) + .with(&update.downstream_id, |sender| sender.clone()) { - if let Err(e) = sender.send(set_difficulty_msg).await { + sender.send(set_difficulty_msg).await.map_err(|e| { error!( "Failed to send SetDifficulty to downstream {}: {:?}", update.downstream_id, e ); - } else { - trace!("Sent SetDifficulty to downstream {}", update.downstream_id); - } + TproxyError::shutdown(TproxyErrorKind::ChannelErrorSender) + })?; + trace!("Sent SetDifficulty to downstream {}", update.downstream_id); } } + Ok(()) } /// Sends an UpdateChannel message for aggregated mode when downstream state changes /// (e.g., disconnect). Calculates total hashrate and minimum target among all remaining /// downstreams. - pub async fn send_update_channel_on_downstream_state_change(&self) { + pub async fn send_update_channel_on_downstream_state_change( + &self, + ) -> TproxyResult<(), error::Sv1Server> { if self.mode.is_non_aggregated() { - return; + return Ok(()); } let is_empty = self.downstreams.is_empty(); @@ -514,41 +554,35 @@ impl Sv1Server { let mut min_target: Option = None; let shares_per_minute = self.shares_per_minute as f64; - for downstream in self.downstreams.iter() { - let downstream_id = *downstream.key(); - let downstream = downstream.value(); - let hashrate = downstream.downstream_data.super_safe_lock(|d| { + self.downstreams.try_for_each(|downstream_id, downstream| { + let hashrate = downstream.downstream_data.with(|d| { d.pending_hashrate.unwrap_or_else(|| { d.hashrate .expect("vardiff implies downstream must have a hashrate") }) - }); + }).map_err(TproxyError::shutdown)?; // UpdateChannel is upstream-facing, so rebuild the exact target from // hashrate instead of reusing the rounded SV1 advertised target. - let target = match hash_rate_to_target(hashrate as f64, shares_per_minute) { - Ok(target) => target, - Err(e) => { - error!( - "Failed to calculate exact target for downstream {} hashrate {}: {:?}", - downstream_id, hashrate, e - ); - continue; - } - }; + let target = hash_rate_to_target(hashrate as f64, shares_per_minute).map_err(|e| { + TproxyError::shutdown(TproxyErrorKind::General(format!( + "failed to calculate exact target for downstream {downstream_id} hashrate {hashrate}: {e:?}" + ))) + })?; total_hashrate += hashrate; min_target = Some(match min_target { Some(current) => current.min(target), None => target, }); - } + Ok(()) + })?; let Some(min_target) = min_target else { warn!( "Skipping aggregated UpdateChannel after downstream state change: no exact downstream target is available" ); - return; + return Ok(()); }; AggregatedSnapshot::Active { @@ -574,16 +608,16 @@ impl Sv1Server { }, }; - if let Err(e) = self - .sv1_server_io + self.sv1_server_io .channel_manager_sender .send((Mining::UpdateChannel(update), None)) .await - { - error!( - "Failed to send UpdateChannel after downstream state change: {:?}", - e - ); - } + .map_err(|e| { + error!( + "Failed to send UpdateChannel after downstream state change: {:?}", + e + ); + TproxyError::shutdown(TproxyErrorKind::ChannelErrorSender) + }) } } diff --git a/miner-apps/translator/src/lib/sv1/sv1_server/downstream_message_handler.rs b/miner-apps/translator/src/lib/sv1/sv1_server/downstream_message_handler.rs index 95493c5d9..50172b2e7 100644 --- a/miner-apps/translator/src/lib/sv1/sv1_server/downstream_message_handler.rs +++ b/miner-apps/translator/src/lib/sv1/sv1_server/downstream_message_handler.rs @@ -10,13 +10,32 @@ use tracing::{debug, info, warn}; use crate::{ error, - sv1::Sv1Server, + sv1::{Downstream, Sv1Server}, utils::{ sv1_worker_name_from_sv1_username, validate_sv1_share, SubmitShareWithChannelId, AGGREGATED_CHANNEL_ID, }, }; +impl Sv1Server { + fn with_registered_downstream_sv1( + &self, + downstream_id: usize, + f: F, + ) -> Result> + where + F: FnOnce(&Downstream) -> Result>, + { + match self + .downstreams + .with(&downstream_id, |downstream| f(downstream)) + { + Some(result) => result, + None => Err(Error::UnknownID(downstream_id as u64)), + } + } +} + // Implements `IsServer` for `Sv1Server` to handle the Sv1 messages. #[cfg_attr(not(test), hotpath::measure_all)] impl IsServer<'static> for Sv1Server { @@ -31,32 +50,35 @@ impl IsServer<'static> for Sv1Server { info!("Received mining.configure from SV1 downstream"); debug!("Downstream {downstream_id}: mining.configure = {}", request); - let Some(downstream) = self.downstreams.get(&downstream_id) else { - return Err(Error::UnknownID(downstream_id as u64)); - }; - - downstream.downstream_data.super_safe_lock(|data| { - data.version_rolling_mask = request - .version_rolling_mask() - .map(|mask| HexU32Be(mask & 0x1FFFE000)); + self.with_registered_downstream_sv1(downstream_id, |downstream| { + downstream + .downstream_data + .with(|data| { + data.version_rolling_mask = request + .version_rolling_mask() + .map(|mask| HexU32Be(mask & 0x1FFFE000)); - data.version_rolling_min_bit = request.version_rolling_min_bit_count(); + data.version_rolling_min_bit = request.version_rolling_min_bit_count(); - debug!( - "Negotiated version_rolling_mask: {:?}", - data.version_rolling_mask - ); + debug!( + "Negotiated version_rolling_mask: {:?}", + data.version_rolling_mask + ); - let params = server_to_client::VersionRollingParams::new( - data.version_rolling_mask.clone().unwrap_or(HexU32Be(0)), - data.version_rolling_min_bit.clone().unwrap_or(HexU32Be(0)), - ) - .expect( - "Invalid version rolling params: \ + let params = server_to_client::VersionRollingParams::new( + data.version_rolling_mask.clone().unwrap_or(HexU32Be(0)), + data.version_rolling_min_bit.clone().unwrap_or(HexU32Be(0)), + ) + .expect( + "Invalid version rolling params: \ automatic mask selection is not supported", - ); + ); - Ok((Some(params), Some(false))) + Ok((Some(params), Some(false))) + }) + .map_err(|_| { + Error::UnexpectedMessage("poisoned downstream data lock".to_string()) + })? }) } @@ -101,15 +123,17 @@ impl IsServer<'static> for Sv1Server { ) -> Result> { let downstream_id = client_id.expect("Downstream id should exist"); - let Some(downstream) = self.downstreams.get(&downstream_id) else { - return Err(Error::UnknownID(downstream_id as u64)); - }; - let job_id = &request.job_id; - let Some(channel_id) = downstream - .downstream_data - .super_safe_lock(|data| data.channel_id) + let Some(channel_id) = + self.with_registered_downstream_sv1(downstream_id, |downstream| { + downstream + .downstream_data + .with(|data| data.channel_id) + .map_err(|_| { + Error::UnexpectedMessage("poisoned downstream data lock".to_string()) + }) + })? else { return Ok(false); }; @@ -125,55 +149,62 @@ impl IsServer<'static> for Sv1Server { let job = self .valid_sv1_jobs - .get(&channel_id) - .and_then(|jobs| find_job(jobs.as_ref())); + .with(&channel_id, |jobs| find_job(jobs.as_ref())) + .flatten(); let Some(job) = job else { return Ok(false); }; - downstream.downstream_data.super_safe_lock(|data| { - let channel_id = match data.channel_id { - Some(id) => id, - None => { - error!( - "Cannot submit share: channel_id is None \ + self.with_registered_downstream_sv1(downstream_id, |downstream| { + downstream + .downstream_data + .with(|data| { + let channel_id = match data.channel_id { + Some(id) => id, + None => { + error!( + "Cannot submit share: channel_id is None \ (waiting for OpenExtendedMiningChannelSuccess)" + ); + return Ok(false); + } + }; + + info!( + "Received mining.submit from SV1 downstream for channel id: {}", + channel_id ); - return Ok(false); - } - }; - - info!( - "Received mining.submit from SV1 downstream for channel id: {}", - channel_id - ); - - let is_valid = validate_sv1_share( - request, - data.target, - data.extranonce1.clone().into(), - data.version_rolling_mask.clone(), - job.clone(), - ) - .unwrap_or(false); - - if !is_valid { - error!("Invalid share for channel id: {}", channel_id); - return Ok(false); - } - - data.pending_share = Some(SubmitShareWithChannelId { - channel_id, - downstream_id, - share: request.clone(), - extranonce: data.extranonce1.clone().into(), - extranonce2_len: data.extranonce2_len, - version_rolling_mask: data.version_rolling_mask.clone(), - job_version: data.last_job_version_field, - }); - - Ok(true) + + let is_valid = validate_sv1_share( + request, + data.target, + data.extranonce1.clone().into(), + data.version_rolling_mask.clone(), + job.clone(), + ) + .unwrap_or(false); + + if !is_valid { + error!("Invalid share for channel id: {}", channel_id); + return Ok(false); + } + + data.pending_share = Some(SubmitShareWithChannelId { + channel_id, + downstream_id, + share: request.clone(), + extranonce: data.extranonce1.clone().into(), + extranonce2_len: data.extranonce2_len, + version_rolling_mask: data.version_rolling_mask.clone(), + job_version: data.last_job_version_field, + }); + + Ok(true) + }) + .map_err(|_| { + Error::UnexpectedMessage("poisoned downstream data lock".to_string()) + })? }) } @@ -185,33 +216,34 @@ impl IsServer<'static> for Sv1Server { /// Checks if a Downstream role is authorized. fn is_authorized(&self, client_id: Option, name: &str) -> Result> { let downstream_id = client_id.expect("Downstream id should exist"); - let Some(downstream) = self.downstreams.get(&downstream_id) else { - return Err(Error::UnknownID(downstream_id as u64)); - }; - let is_authorized = downstream - .downstream_data - .super_safe_lock(|data| data.sv1_username == *name); + let is_authorized = self.with_registered_downstream_sv1(downstream_id, |downstream| { + downstream + .downstream_data + .with(|data| data.sv1_username == *name) + .map_err(|_| Error::UnexpectedMessage("poisoned downstream data lock".to_string())) + })?; Ok(is_authorized) } /// Authorizes a Downstream role. fn authorize(&mut self, client_id: Option, name: &str) -> Result<(), Error<'static>> { let downstream_id = client_id.expect("Downstream id should exist"); - let Some(downstream) = self.downstreams.get(&downstream_id) else { - return Err(Error::UnknownID(downstream_id as u64)); - }; - let is_authorized = self.is_authorized(client_id, name)?; - downstream.downstream_data.super_safe_lock(|data| { - if !is_authorized { - data.sv1_username = name.to_string(); - } - data.sv1_worker_name = sv1_worker_name_from_sv1_username(name).to_string(); - debug!( - "Down: Set sv1_username '{}' and sv1_worker_name '{}' for downstream {}", - data.sv1_username, data.sv1_worker_name, downstream_id - ); - }); + self.with_registered_downstream_sv1(downstream_id, |downstream| { + downstream + .downstream_data + .with(|data| { + if !is_authorized { + data.sv1_username = name.to_string(); + } + data.sv1_worker_name = sv1_worker_name_from_sv1_username(name).to_string(); + debug!( + "Down: Set sv1_username '{}' and sv1_worker_name '{}' for downstream {}", + data.sv1_username, data.sv1_worker_name, downstream_id + ); + }) + .map_err(|_| Error::UnexpectedMessage("poisoned downstream data lock".to_string())) + })?; Ok(()) } @@ -223,23 +255,27 @@ impl IsServer<'static> for Sv1Server { _extranonce1: Option>, ) -> Result, Error<'static>> { let downstream_id = client_id.expect("Downstream id should exist"); - let Some(downstream) = self.downstreams.get(&downstream_id) else { - return Err(Error::UnknownID(downstream_id as u64)); - }; - downstream - .downstream_data - .super_safe_lock(|data| Ok(data.extranonce1.clone())) + self.with_registered_downstream_sv1(downstream_id, |downstream| { + downstream + .downstream_data + .with(|data| Ok(data.extranonce1.clone())) + .map_err(|_| { + Error::UnexpectedMessage("poisoned downstream data lock".to_string()) + })? + }) } /// Returns the `Downstream`'s `extranonce1` value. fn extranonce1(&self, client_id: Option) -> Result, Error<'static>> { let downstream_id = client_id.expect("Downstream id should exist"); - let Some(downstream) = self.downstreams.get(&downstream_id) else { - return Err(Error::UnknownID(downstream_id as u64)); - }; - downstream - .downstream_data - .super_safe_lock(|data| Ok(data.extranonce1.clone())) + self.with_registered_downstream_sv1(downstream_id, |downstream| { + downstream + .downstream_data + .with(|data| Ok(data.extranonce1.clone())) + .map_err(|_| { + Error::UnexpectedMessage("poisoned downstream data lock".to_string()) + })? + }) } /// Sets the `extranonce2_size` field sent in the SV1 `mining.notify` message to the value @@ -250,23 +286,27 @@ impl IsServer<'static> for Sv1Server { _extra_nonce2_size: Option, ) -> Result> { let downstream_id = client_id.expect("Downstream id should exist"); - let Some(downstream) = self.downstreams.get(&downstream_id) else { - return Err(Error::UnknownID(downstream_id as u64)); - }; - downstream - .downstream_data - .super_safe_lock(|data| Ok(data.extranonce2_len)) + self.with_registered_downstream_sv1(downstream_id, |downstream| { + downstream + .downstream_data + .with(|data| Ok(data.extranonce2_len)) + .map_err(|_| { + Error::UnexpectedMessage("poisoned downstream data lock".to_string()) + })? + }) } /// Returns the `Downstream`'s `extranonce2_size` value. fn extranonce2_size(&self, client_id: Option) -> Result> { let downstream_id = client_id.expect("Downstream id should exist"); - let Some(downstream) = self.downstreams.get(&downstream_id) else { - return Err(Error::UnknownID(downstream_id as u64)); - }; - downstream - .downstream_data - .super_safe_lock(|data| Ok(data.extranonce2_len)) + self.with_registered_downstream_sv1(downstream_id, |downstream| { + downstream + .downstream_data + .with(|data| Ok(data.extranonce2_len)) + .map_err(|_| { + Error::UnexpectedMessage("poisoned downstream data lock".to_string()) + })? + }) } /// Returns the version rolling mask. @@ -275,12 +315,14 @@ impl IsServer<'static> for Sv1Server { client_id: Option, ) -> Result, Error<'static>> { let downstream_id = client_id.expect("Downstream id should exist"); - let Some(downstream) = self.downstreams.get(&downstream_id) else { - return Err(Error::UnknownID(downstream_id as u64)); - }; - downstream - .downstream_data - .super_safe_lock(|data| Ok(data.version_rolling_mask.clone())) + self.with_registered_downstream_sv1(downstream_id, |downstream| { + downstream + .downstream_data + .with(|data| Ok(data.version_rolling_mask.clone())) + .map_err(|_| { + Error::UnexpectedMessage("poisoned downstream data lock".to_string()) + })? + }) } /// Sets the version rolling mask. @@ -290,13 +332,12 @@ impl IsServer<'static> for Sv1Server { mask: Option, ) -> Result<(), Error<'static>> { let downstream_id = client_id.expect("Downstream id should exist"); - let Some(downstream) = self.downstreams.get(&downstream_id) else { - return Err(Error::UnknownID(downstream_id as u64)); - }; - - downstream - .downstream_data - .super_safe_lock(|data| data.version_rolling_mask = mask); + self.with_registered_downstream_sv1(downstream_id, |downstream| { + downstream + .downstream_data + .with(|data| data.version_rolling_mask = mask) + .map_err(|_| Error::UnexpectedMessage("poisoned downstream data lock".to_string())) + })?; Ok(()) } @@ -308,12 +349,12 @@ impl IsServer<'static> for Sv1Server { mask: Option, ) -> Result<(), Error<'static>> { let downstream_id = client_id.expect("Downstream id should exist"); - let Some(downstream) = self.downstreams.get(&downstream_id) else { - return Err(Error::UnknownID(downstream_id as u64)); - }; - downstream - .downstream_data - .super_safe_lock(|data| data.version_rolling_min_bit = mask); + self.with_registered_downstream_sv1(downstream_id, |downstream| { + downstream + .downstream_data + .with(|data| data.version_rolling_min_bit = mask) + .map_err(|_| Error::UnexpectedMessage("poisoned downstream data lock".to_string())) + })?; Ok(()) } diff --git a/miner-apps/translator/src/lib/sv1/sv1_server/mod.rs b/miner-apps/translator/src/lib/sv1/sv1_server/mod.rs index fa2c9f558..0617911fd 100644 --- a/miner-apps/translator/src/lib/sv1/sv1_server/mod.rs +++ b/miner-apps/translator/src/lib/sv1/sv1_server/mod.rs @@ -28,11 +28,9 @@ use crate::{ }, }; use async_channel::{unbounded, Receiver, Sender}; -use dashmap::DashMap; #[cfg(feature = "monitoring")] use std::net::IpAddr; use std::{ - collections::HashMap, net::SocketAddr, sync::{ atomic::{AtomicU32, AtomicUsize, Ordering}, @@ -44,7 +42,6 @@ use std::{ use stratum_apps::monitoring::{MinerTelemetry, MinerTelemetryStatus}; use stratum_apps::{ channel_utils::ReceiverCleanup, - custom_mutex::Mutex, fallback_coordinator::FallbackCoordinator, network_helpers::sv1_connection::ConnectionSV1, stratum_core::{ @@ -69,6 +66,7 @@ use stratum_apps::{ }, sv1_api::{json_rpc, server_to_client, utils::HexU32Be, IsServer}, }, + sync::{SharedLock, SharedMap}, task_manager::TaskManager, utils::types::{ChannelId, DownstreamId, Hashrate, RequestId, SharesPerMinute}, }; @@ -80,7 +78,7 @@ const SV1_MIN_DIFFICULTY_FOR_INTEGER_POWER_OF_TWO_ROUNDING: f64 = 1.0; #[derive(Clone)] struct Sv1ServerIo { - sv1_server_to_downstream_sender: Arc>>>, + sv1_server_to_downstream_sender: SharedMap>, downstream_to_sv1_server_sender: Sender<(DownstreamId, json_rpc::Message)>, downstream_to_sv1_server_receiver: Receiver<(DownstreamId, json_rpc::Message)>, channel_manager_receiver: Receiver>, @@ -97,7 +95,7 @@ impl Sv1ServerIo { let (downstream_to_sv1_server_sender, downstream_to_sv1_server_receiver) = unbounded(); Self { - sv1_server_to_downstream_sender: Arc::new(Mutex::new(HashMap::new())), + sv1_server_to_downstream_sender: SharedMap::new(), downstream_to_sv1_server_receiver, downstream_to_sv1_server_sender, channel_manager_receiver, @@ -111,12 +109,8 @@ impl Sv1ServerIo { self.channel_manager_receiver.close_and_drain(); self.downstream_to_sv1_server_receiver.close_and_drain(); self.sv1_server_to_downstream_sender - .super_safe_lock(|downstream_senders| { - for sender in downstream_senders.values() { - sender.close(); - } - downstream_senders.clear(); - }); + .for_each(|_, sender| sender.close()); + self.sv1_server_to_downstream_sender.clear(); } } @@ -124,20 +118,20 @@ impl Sv1ServerIo { #[derive(Clone)] pub(crate) struct MinerTelemetryState { /// Latest telemetry fetched for each matched SV1 downstream connection. - pub(crate) telemetry: Arc>, + pub(crate) telemetry: SharedMap, /// Miner management IP selected for each matched SV1 downstream connection. - pub(crate) management_ips: Arc>, + pub(crate) management_ips: SharedMap, /// Latest telemetry matching status for each active SV1 downstream connection. - pub(crate) statuses: Arc>, + pub(crate) statuses: SharedMap, } #[cfg(feature = "monitoring")] impl MinerTelemetryState { fn new() -> Self { Self { - telemetry: Arc::new(DashMap::new()), - management_ips: Arc::new(DashMap::new()), - statuses: Arc::new(DashMap::new()), + telemetry: SharedMap::new(), + management_ips: SharedMap::new(), + statuses: SharedMap::new(), } } @@ -155,16 +149,16 @@ impl MinerTelemetryState { pub(crate) fn telemetry_for(&self, downstream_id: DownstreamId) -> Option { self.telemetry - .get(&downstream_id) - .map(|telemetry| telemetry.clone()) + .with(&downstream_id, |telemetry| telemetry.clone()) } pub(crate) fn management_ip_for(&self, downstream_id: DownstreamId) -> Option { - self.management_ips.get(&downstream_id).map(|ip| *ip) + self.management_ips + .with(&downstream_id, |management_ip| *management_ip) } pub(crate) fn status_for(&self, downstream_id: DownstreamId) -> Option { - self.statuses.get(&downstream_id).map(|status| *status) + self.statuses.with(&downstream_id, |status| *status) } } @@ -189,20 +183,20 @@ pub struct Sv1Server { pub(crate) keepalive_job_id_counter: Arc, pub(crate) downstream_id_factory: Arc, pub(crate) request_id_factory: Arc, - pub(crate) downstreams: Arc>, + pub(crate) downstreams: SharedMap, #[cfg(feature = "monitoring")] pub(crate) miner_telemetry: MinerTelemetryState, - pub(crate) request_id_to_downstream_id: Arc>, - pub(crate) channel_id_to_downstream_id: Arc>>, - pub(crate) vardiff: Arc>>>, + pub(crate) request_id_to_downstream_id: SharedMap, + pub(crate) channel_id_to_downstream_id: SharedMap, + pub(crate) vardiff: SharedMap, /// HashMap to store the SetNewPrevHash for each channel /// Used in both aggregated and non-aggregated mode - pub(crate) prevhashes: Arc>>, + pub(crate) prevhashes: SharedMap>, /// Tracks pending target updates that are waiting for SetTarget response from upstream - pub(crate) pending_target_updates: Arc>>, + pub(crate) pending_target_updates: SharedLock>, /// Valid Sv1 jobs storage, containing only a single shared entry (AGGREGATED_CHANNEL_ID) in /// case of channels aggregation (aggregated mode) - pub(crate) valid_sv1_jobs: Arc>>>, + pub(crate) valid_sv1_jobs: SharedMap>>, pub(crate) mode: TproxyMode, user_identity: Arc>, } @@ -278,10 +272,12 @@ impl Sv1Server { msg: stratum_apps::stratum_core::sv1_api::json_rpc::Message, ) { if channel_id == AGGREGATED_CHANNEL_ID { - let downstream_senders = self - .sv1_server_io + let mut downstream_senders = Vec::new(); + self.sv1_server_io .sv1_server_to_downstream_sender - .super_safe_lock(|downstream_channels| downstream_channels.clone()); + .for_each(|downstream_id, sender| { + downstream_senders.push((downstream_id, sender.clone())); + }); // Broadcast to every connected downstream. for (downstream_id, sender) in downstream_senders { if let Err(e) = sender.send(msg.clone()).await { @@ -295,7 +291,7 @@ impl Sv1Server { // Non-aggregated: send to the single downstream that owns this channel_id. let downstream_id = match self .channel_id_to_downstream_id - .super_safe_lock(|map| map.get(&channel_id).cloned()) + .with(&channel_id, |downstream_id| *downstream_id) { Some(id) => id, None => return, @@ -304,7 +300,7 @@ impl Sv1Server { let sender = self .sv1_server_io .sv1_server_to_downstream_sender - .super_safe_lock(|ch| ch.get(&downstream_id).cloned()); + .with(&downstream_id, |sender| sender.clone()); let Some(sender) = sender else { return }; @@ -327,15 +323,40 @@ impl Sv1Server { self.downstreams.clear(); #[cfg(feature = "monitoring")] self.miner_telemetry.clear(); - self.channel_id_to_downstream_id - .super_safe_lock(|map| map.clear()); + self.channel_id_to_downstream_id.clear(); self.request_id_to_downstream_id.clear(); self.pending_target_updates - .safe_lock(|updates| updates.clear()) + .with(|updates| updates.clear()) .ok(); self.sv1_server_io.close(); } + /// Runs `f` while holding the downstream map entry guard. + /// + /// Use this when mutations must only happen if the downstream is still + /// registered in Sv1Server. Keep `f` short: do not perform blocking work, + /// send messages, await, or re-enter `self.downstreams` inside it. + #[allow(clippy::result_large_err)] + pub(crate) fn with_registered_downstream( + &self, + downstream_id: DownstreamId, + f: F, + ) -> TproxyResult + where + F: FnOnce(&Downstream) -> TproxyResult, + { + match self + .downstreams + .with(&downstream_id, |downstream| f(downstream)) + { + Some(result) => result, + None => Err(TproxyError::disconnect( + TproxyErrorKind::DownstreamNotFound(downstream_id as u32), + downstream_id, + )), + } + } + /// Creates a new SV1 server instance. /// /// # Arguments @@ -365,15 +386,15 @@ impl Sv1Server { keepalive_job_id_counter: Arc::new(AtomicU32::new(0)), downstream_id_factory: Arc::new(AtomicUsize::new(1)), request_id_factory: Arc::new(AtomicU32::new(1)), - downstreams: Arc::new(DashMap::new()), + downstreams: SharedMap::new(), #[cfg(feature = "monitoring")] miner_telemetry: MinerTelemetryState::new(), - request_id_to_downstream_id: Arc::new(DashMap::new()), - channel_id_to_downstream_id: Arc::new(Mutex::new(HashMap::new())), - vardiff: Arc::new(DashMap::new()), - prevhashes: Arc::new(DashMap::new()), - pending_target_updates: Arc::new(Mutex::new(Vec::new())), - valid_sv1_jobs: Arc::new(DashMap::new()), + request_id_to_downstream_id: SharedMap::new(), + channel_id_to_downstream_id: SharedMap::new(), + vardiff: SharedMap::new(), + prevhashes: SharedMap::new(), + pending_target_updates: SharedLock::new(Vec::new()), + valid_sv1_jobs: SharedMap::new(), mode, user_identity: Arc::new(OnceLock::new()), } @@ -484,7 +505,9 @@ impl Sv1Server { ).await; let downstream_id = self.downstream_id_factory.fetch_add(1, Ordering::Relaxed); let (sv1_server_sender, sv1_server_receiver) = async_channel::unbounded(); - self.sv1_server_io.sv1_server_to_downstream_sender.super_safe_lock(|map| map.insert(downstream_id, sv1_server_sender)); + self.sv1_server_io + .sv1_server_to_downstream_sender + .insert(downstream_id, sv1_server_sender); let downstream = Downstream::new( downstream_id, @@ -503,7 +526,7 @@ impl Sv1Server { // Insert vardiff state for this downstream only if vardiff is enabled if self.config.downstream_difficulty_config.enable_vardiff { let vardiff = VardiffState::new().expect("Failed to create vardiffstate"); - self.vardiff.insert(downstream_id, Arc::new(Mutex::new(vardiff))); + self.vardiff.insert(downstream_id, vardiff); } info!("Downstream {} registered successfully (channel will be opened after first message)", downstream_id); @@ -551,7 +574,19 @@ impl Sv1Server { } } } - _ = &mut vardiff_future, if vardiff_enabled => {} + res = &mut vardiff_future, if vardiff_enabled => { + if let Err(e) = res { + if let LoopControl::Break = self.handle_error_action( + "Sv1Server::spawn_vardiff_loop", + &e, + &cancellation_token, + &fallback_token, + ).await { + self.cleanup(); + break; + } + } + } _ = &mut keepalive_future, if keepalive_enabled => {} } } @@ -583,21 +618,23 @@ impl Sv1Server { .await .map_err(TproxyError::shutdown)?; - let Some(downstream) = self - .downstreams - .get(&downstream_id) - .map(|r| r.value().clone()) - else { - return Ok(()); + let channel_id = match self.with_registered_downstream(downstream_id, |downstream| { + downstream + .downstream_data + .with(|data| data.channel_id) + .map_err(TproxyError::shutdown) + }) { + Ok(channel_id) => channel_id, + Err(_) => return Ok(()), }; - - let channel_id = downstream - .downstream_data - .super_safe_lock(|data| data.channel_id); if channel_id.is_none() { - let is_first_message = downstream - .downstream_data - .super_safe_lock(|d| d.queued_sv1_handshake_messages.is_empty()); + let is_first_message = + self.with_registered_downstream(downstream_id, |downstream| { + downstream + .downstream_data + .with(|d| d.queued_sv1_handshake_messages.is_empty()) + .map_err(TproxyError::shutdown) + })?; if is_first_message { self.handle_open_channel_request(downstream_id).await?; debug!( @@ -606,10 +643,15 @@ impl Sv1Server { ); } debug!("Down: Queuing Sv1 message until channel is established"); - downstream.downstream_data.super_safe_lock(|data| { - data.queued_sv1_handshake_messages - .push(downstream_message.clone()) - }); + self.with_registered_downstream(downstream_id, |downstream| { + downstream + .downstream_data + .with(|data| { + data.queued_sv1_handshake_messages + .push(downstream_message.clone()) + }) + .map_err(TproxyError::shutdown) + })?; return Ok(()); } @@ -622,9 +664,14 @@ impl Sv1Server { match response { Ok(Some(response_msg)) => { debug!("Down: Sending Sv1 message to downstream: {}", response_msg); - downstream - .downstream_io - .downstream_sv1_sender + let (downstream_sv1_sender, downstream) = + self.with_registered_downstream(downstream_id, |downstream| { + Ok(( + downstream.downstream_io.downstream_sv1_sender.clone(), + downstream.clone(), + )) + })?; + downstream_sv1_sender .send(response_msg.into()) .await .map_err(|error| { @@ -651,9 +698,12 @@ impl Sv1Server { } // Check if there's a pending share to send to the Sv1Server - let pending_share = downstream - .downstream_data - .super_safe_lock(|d| d.pending_share.take()); + let pending_share = self.with_registered_downstream(downstream_id, |downstream| { + downstream + .downstream_data + .with(|d| d.pending_share.take()) + .map_err(TproxyError::shutdown) + })?; if let Some(share) = pending_share { self.handle_submit_shares(share).await?; } @@ -668,9 +718,9 @@ impl Sv1Server { ) -> TproxyResult<(), error::Sv1Server> { // Increment vardiff counter for this downstream (only if vardiff is enabled) if self.config.downstream_difficulty_config.enable_vardiff { - if let Some(vardiff_state) = self.vardiff.get(&message.downstream_id) { - vardiff_state.super_safe_lock(|state| state.increment_shares_since_last_update()); - } + self.vardiff.with_mut(&message.downstream_id, |state| { + state.increment_shares_since_last_update(); + }); } let job_version = match message.job_version { @@ -712,24 +762,13 @@ impl Sv1Server { ) .map_err(|_| TproxyError::shutdown(TproxyErrorKind::SV1Error))?; - let Some(downstream) = self - .downstreams - .get(&message.downstream_id) - .map(|r| r.value().clone()) - else { - warn!( - "Downstream {} disconnected before share could be submitted, dropping share", - message.downstream_id - ); - return Ok(()); - }; - let sv1_worker_name = downstream.downstream_data.super_safe_lock(|d| { - if d.sv1_worker_name.is_empty() { - None - } else { - Some(d.sv1_worker_name.clone()) - } - }); + let worker_name = self.with_registered_downstream(message.downstream_id, |downstream| { + downstream + .downstream_data + .with(|data| data.sv1_worker_name.clone()) + .map_err(TproxyError::shutdown) + })?; + let sv1_worker_name = (!worker_name.is_empty()).then_some(worker_name); self.sv1_server_io .channel_manager_sender @@ -812,50 +851,63 @@ impl Sv1Server { m.request_id, ))); }; - if let Some(downstream) = self.downstreams.get(&downstream_id) { - let initial_target = Target::from_le_bytes(m.target.to_array()); - let extranonce1 = m - .extranonce_prefix - .to_owned_bytes() - .try_into() - .map_err(TproxyError::fallback)?; - downstream - .downstream_data - .safe_lock(|d| { - d.extranonce1 = extranonce1; - d.extranonce2_len = m.extranonce_size.into(); - d.channel_id = Some(m.channel_id); - // Set the initial upstream target from OpenExtendedMiningChannelSuccess - d.set_upstream_target(initial_target, downstream_id); - }) - .map_err(TproxyError::shutdown)?; - self.channel_id_to_downstream_id - .super_safe_lock(|map| map.insert(m.channel_id, downstream_id)); - - // Process all queued messages now that channel is established - let queued_messages = downstream.downstream_data.super_safe_lock(|d| { - let messages = d.queued_sv1_handshake_messages.clone(); - d.queued_sv1_handshake_messages.clear(); - messages + let initial_target = Target::from_le_bytes(m.target.to_array()); + let extranonce1 = m + .extranonce_prefix + .to_owned_bytes() + .try_into() + .map_err(TproxyError::fallback)?; + let downstream_setup = + self.with_registered_downstream(downstream_id, |downstream| { + downstream + .downstream_data + .with(|d| { + d.extranonce1 = extranonce1; + d.extranonce2_len = m.extranonce_size.into(); + d.channel_id = Some(m.channel_id); + // Set the initial upstream target from + // OpenExtendedMiningChannelSuccess + d.set_upstream_target(initial_target, downstream_id); + }) + .map_err(TproxyError::shutdown)?; + + let queued_messages = downstream + .downstream_data + .with(|d| { + let messages = d.queued_sv1_handshake_messages.clone(); + d.queued_sv1_handshake_messages.clear(); + messages + }) + .map_err(TproxyError::shutdown)?; + + Ok(( + queued_messages, + downstream.downstream_io.downstream_sv1_sender.clone(), + downstream.clone(), + )) }); - { - if !queued_messages.is_empty() { - info!( - "Processing {} queued Sv1 messages for downstream {}", - queued_messages.len(), - downstream_id - ); - - let downstream_sv1_sender = - downstream.downstream_io.downstream_sv1_sender.clone(); - - for message in queued_messages { - let is_authorize = is_mining_authorize(&message); - let response = - self.clone().handle_message(Some(downstream_id), message); - match response { - Ok(Some(response_msg)) => { - downstream_sv1_sender.send(response_msg.into()).await + + match downstream_setup { + Ok((queued_messages, downstream_sv1_sender, downstream)) => { + self.channel_id_to_downstream_id + .insert(m.channel_id, downstream_id); + + // Process all queued messages now that channel is established + { + if !queued_messages.is_empty() { + info!( + "Processing {} queued Sv1 messages for downstream {}", + queued_messages.len(), + downstream_id + ); + + for message in queued_messages { + let is_authorize = is_mining_authorize(&message); + let response = + self.clone().handle_message(Some(downstream_id), message); + match response { + Ok(Some(response_msg)) => { + downstream_sv1_sender.send(response_msg.into()).await .map_err(|e| { error!( "Down: Failed to send message to downstream: {e:?}" @@ -865,55 +917,65 @@ impl Sv1Server { ) })?; - if is_authorize { - info!("Down: Handling mining.authorize after upstream channel is open"); - if let Err(e) = - downstream.handle_sv1_handshake_completion().await - { - error!( + if is_authorize { + info!("Down: Handling mining.authorize after upstream channel is open"); + if let Err(e) = downstream + .handle_sv1_handshake_completion() + .await + { + error!( "Down: Failed to handle handshake completion: {:?}", e ); - return Err(TproxyError::disconnect( - e, - downstream_id, - )); + return Err(TproxyError::disconnect( + e, + downstream_id, + )); + } } } - } - Ok(None) => { - // Message was handled but no response needed - } - Err(e) => { - error!("Down: Error handling downstream message: {:?}", e); - return Err(TproxyError::disconnect(e, downstream_id)); + Ok(None) => { + // Message was handled but no response needed + } + Err(e) => { + error!( + "Down: Error handling downstream message: {:?}", + e + ); + return Err(TproxyError::disconnect(e, downstream_id)); + } } } } } - } - let set_difficulty = + let set_difficulty = build_sv1_set_difficulty_from_sv2_target_with_integer_power_of_two_rounding( first_target, SV1_MIN_DIFFICULTY_FOR_INTEGER_POWER_OF_TWO_ROUNDING, ) .map_err(TproxyError::shutdown)?; - // send the set_difficulty message to the downstream - if let Some(sender) = self - .sv1_server_io - .sv1_server_to_downstream_sender - .super_safe_lock(|map| map.get(&downstream_id).cloned()) - { - sender.send(set_difficulty).await.map_err(|_| { - TproxyError::disconnect( - TproxyErrorKind::ChannelErrorSender, - downstream_id, - ) - })?; + // send the set_difficulty message to the downstream + if let Some(sender) = self + .sv1_server_io + .sv1_server_to_downstream_sender + .with(&downstream_id, |sender| sender.clone()) + { + sender.send(set_difficulty).await.map_err(|_| { + TproxyError::disconnect( + TproxyErrorKind::ChannelErrorSender, + downstream_id, + ) + })?; + } + } + Err(e) => { + if matches!(e.kind, TproxyErrorKind::DownstreamNotFound(_)) { + error!("Downstream not found for downstream_id: {}", downstream_id); + } else { + return Err(e); + } } - } else { - error!("Downstream not found for downstream_id: {}", downstream_id); } } @@ -922,11 +984,10 @@ impl Sv1Server { "Received NewExtendedMiningJob for channel id: {}", m.channel_id ); - // Clone the prevhash immediately so the DashMap guard is not held across .await. + // Clone the prevhash immediately so shared map access is not held across .await. if let Some(prevhash) = self .prevhashes - .get(&m.channel_id) - .map(|r| r.value().clone()) + .with(&m.channel_id, |prevhash| prevhash.clone()) { let prevhash = prevhash.as_static(); let clean_jobs = m.job_id == prevhash.job_id; @@ -942,14 +1003,13 @@ impl Sv1Server { AGGREGATED_CHANNEL_ID }; - { - let mut channel_jobs = - self.valid_sv1_jobs.entry(job_channel_id).or_default(); - if clean_jobs { - channel_jobs.clear(); - } - channel_jobs.push(notify_parsed); - } + self.valid_sv1_jobs + .with_mut_or_default(job_channel_id, |channel_jobs| { + if clean_jobs { + channel_jobs.clear(); + } + channel_jobs.push(notify_parsed); + }); let notify_msg: stratum_apps::stratum_core::sv1_api::json_rpc::Message = notify.into(); @@ -967,7 +1027,7 @@ impl Sv1Server { debug!("Received SetTarget for channel id: {}", m.channel_id); if self.config.downstream_difficulty_config.enable_vardiff { // Vardiff enabled - use full difficulty management - self.handle_set_target_message(m).await; + self.handle_set_target_message(m).await?; } else { // Vardiff disabled - just forward the difficulty to downstreams debug!("Vardiff disabled - forwarding SetTarget to downstreams"); @@ -1073,7 +1133,7 @@ impl Sv1Server { self.miner_telemetry.remove_downstream(downstream_id); self.sv1_server_io .sv1_server_to_downstream_sender - .super_safe_lock(|map| map.remove(&downstream_id)); + .remove(&downstream_id); let current_downstream = self.downstreams.remove(&downstream_id); @@ -1082,13 +1142,21 @@ impl Sv1Server { // In aggregated mode, send UpdateChannel to reflect the new state (only if vardiff // enabled) if self.config.downstream_difficulty_config.enable_vardiff { - self.send_update_channel_on_downstream_state_change().await; + if let Err(e) = self.send_update_channel_on_downstream_state_change().await { + error!( + "Failed to send UpdateChannel after downstream {} disconnect: {:?}", + downstream_id, e + ); + } } - let channel_id = downstream.downstream_data.super_safe_lock(|d| d.channel_id); + let channel_id = downstream + .downstream_data + .with(|d| d.channel_id) + .ok() + .flatten(); if let Some(channel_id) = channel_id { - self.channel_id_to_downstream_id - .super_safe_lock(|map| map.remove(&channel_id)); + self.channel_id_to_downstream_id.remove(&channel_id); // Send `CloseChannel` to the channel manager in both modes so // it can free the per-downstream `ExtendedChannel` (and, in // aggregated mode, the allocator-minted `ExtranoncePrefix` @@ -1176,12 +1244,11 @@ impl Sv1Server { target: Target, derived_hashrate: Option, ) -> TproxyResult<(), error::Sv1Server> { - let tasks: Vec<(DownstreamId, _)> = self - .downstreams - .iter() - .filter_map(|entry| { - let downstream_id = *entry.key(); - let has_channel = entry.value().downstream_data.super_safe_lock(|d| { + let mut tasks = Vec::new(); + self.downstreams.for_each(|downstream_id, downstream| { + let has_channel = downstream + .downstream_data + .with(|d| { let channel_id = d.channel_id?; d.set_upstream_target(target, downstream_id); // Downstream validation must use the advertised (pow2 @@ -1199,21 +1266,24 @@ impl Sv1Server { d.set_pending_hashrate(Some(hr as f32), downstream_id); } Some(channel_id) - }); - if has_channel.is_none() { - trace!( - "Skipping downstream {}: no channel_id set (vardiff disabled)", - downstream_id - ); - return None; - } - let sender = self - .sv1_server_io - .sv1_server_to_downstream_sender - .super_safe_lock(|map| map.get(&downstream_id).cloned())?; - Some((downstream_id, sender)) - }) - .collect(); + }) + .ok() + .flatten(); + if has_channel.is_none() { + trace!( + "Skipping downstream {}: no channel_id set (vardiff disabled)", + downstream_id + ); + return; + } + if let Some(sender) = self + .sv1_server_io + .sv1_server_to_downstream_sender + .with(&downstream_id, |sender| sender.clone()) + { + tasks.push((downstream_id, sender)); + } + }); for (downstream_id, sender) in tasks { let set_difficulty_msg = @@ -1257,7 +1327,7 @@ impl Sv1Server { ) -> TproxyResult<(), error::Sv1Server> { let Some(downstream_id) = self .channel_id_to_downstream_id - .super_safe_lock(|map| map.get(&channel_id).cloned()) + .with(&channel_id, |downstream_id| *downstream_id) else { warn!( "No downstream found for channel {} when vardiff is disabled", @@ -1281,26 +1351,33 @@ impl Sv1Server { )); }; - let Some(downstream) = self.downstreams.get(&downstream_id) else { - return Ok(()); - }; - downstream.downstream_data.super_safe_lock(|d| { - d.set_upstream_target(target, downstream_id); - // See send_set_difficulty_to_all_downstreams: downstream validation - // uses the advertised pow2 difficulty. - d.set_pending_target( - sv1_advertised_target_from_sv2_target( - target, - SV1_MIN_DIFFICULTY_FOR_INTEGER_POWER_OF_TWO_ROUNDING, - ) - .unwrap_or(target), - downstream_id, - ); - // Update pending hashrate derived from the upstream target - if let Some(hr) = derived_hashrate { - d.set_pending_hashrate(Some(hr as f32), downstream_id); + if let Err(e) = self.with_registered_downstream(downstream_id, |downstream| { + downstream + .downstream_data + .with(|d| { + d.set_upstream_target(target, downstream_id); + // See send_set_difficulty_to_all_downstreams: downstream validation + // uses the advertised pow2 difficulty. + d.set_pending_target( + sv1_advertised_target_from_sv2_target( + target, + SV1_MIN_DIFFICULTY_FOR_INTEGER_POWER_OF_TWO_ROUNDING, + ) + .unwrap_or(target), + downstream_id, + ); + // Update pending hashrate derived from the upstream target + if let Some(hr) = derived_hashrate { + d.set_pending_hashrate(Some(hr as f32), downstream_id); + } + }) + .map_err(TproxyError::shutdown) + }) { + if matches!(e.kind, TproxyErrorKind::DownstreamNotFound(_)) { + return Ok(()); } - }); + return Err(e); + } let set_difficulty_msg = match build_sv1_set_difficulty_from_sv2_target_with_integer_power_of_two_rounding( @@ -1320,7 +1397,7 @@ impl Sv1Server { let sender = self .sv1_server_io .sv1_server_to_downstream_sender - .super_safe_lock(|map| map.get(&downstream_id).cloned()); + .with(&downstream_id, |sender| sender.clone()); if let Some(sender) = sender { if let Err(e) = sender.send(set_difficulty_msg).await { @@ -1359,13 +1436,11 @@ impl Sv1Server { loop { tokio::time::sleep(check_interval).await; - let keepalive_targets: Vec<(DownstreamId, Option)> = self - .downstreams - .iter() - .filter_map(|downstream| { - let downstream_id = downstream.key(); - let downstream = downstream.value(); - downstream.downstream_data.super_safe_lock(|d| { + let mut keepalive_targets = Vec::new(); + self.downstreams.for_each(|downstream_id, downstream| { + let keepalive_target = downstream + .downstream_data + .with(|d| { // Only send keepalive if: // 1. Handshake is complete // 2. Enough time has passed since last job @@ -1382,13 +1457,17 @@ impl Sv1Server { }; if needs_keepalive { - Some((*downstream_id, d.channel_id)) + Some((downstream_id, d.channel_id)) } else { None } }) - }) - .collect(); + .ok() + .flatten(); + if let Some(keepalive_target) = keepalive_target { + keepalive_targets.push(keepalive_target); + } + }); // Send keepalive to each downstream that needs one for (downstream_id, channel_id) in keepalive_targets { @@ -1436,9 +1515,10 @@ impl Sv1Server { channel_id }; - _ = job_channel_id - .and_then(|ch_id| self.valid_sv1_jobs.get_mut(&ch_id)) - .map(|mut jobs| jobs.push(keepalive_notify.clone())); + if let Some(ch_id) = job_channel_id { + self.valid_sv1_jobs + .with_mut_or_default(ch_id, |jobs| jobs.push(keepalive_notify.clone())); + } Some(keepalive_notify) }); @@ -1452,7 +1532,7 @@ impl Sv1Server { let sent = match self .sv1_server_io .sv1_server_to_downstream_sender - .super_safe_lock(|map| map.get(&downstream_id).cloned()) + .with(&downstream_id, |sender| sender.clone()) { Some(sender) => sender.send(notify.into()).await.is_ok(), None => false, @@ -1462,9 +1542,14 @@ impl Sv1Server { "Failed to send keepalive job to downstream {}", downstream_id ); - } else if let Some(downstream) = self.downstreams.get(&downstream_id) { - downstream.downstream_data.super_safe_lock(|d| { - d.last_job_received_time = Some(Instant::now()); + } else { + _ = self.with_registered_downstream(downstream_id, |downstream| { + downstream + .downstream_data + .with(|d| { + d.last_job_received_time = Some(Instant::now()); + }) + .map_err(TproxyError::shutdown) }); } } @@ -1507,8 +1592,8 @@ impl Sv1Server { }; self.valid_sv1_jobs - .get(&channel_id) - .and_then(|jobs| jobs.last().cloned()) + .with(&channel_id, |jobs| jobs.last().cloned()) + .flatten() } /// Gets the original upstream job by its job_id. @@ -1525,10 +1610,10 @@ impl Sv1Server { }; self.valid_sv1_jobs - .get(&channel_id)? - .iter() - .find(|j| j.job_id == job_id) - .cloned() + .with(&channel_id, |jobs| { + jobs.iter().find(|j| j.job_id == job_id).cloned() + }) + .flatten() } } diff --git a/miner-apps/translator/src/lib/sv2/channel_manager/extensions_message_handler.rs b/miner-apps/translator/src/lib/sv2/channel_manager/extensions_message_handler.rs index 3b9188713..0aee9664c 100644 --- a/miner-apps/translator/src/lib/sv2/channel_manager/extensions_message_handler.rs +++ b/miner-apps/translator/src/lib/sv2/channel_manager/extensions_message_handler.rs @@ -21,9 +21,9 @@ impl HandleExtensionsFromServerAsync for ChannelManager { &self, _server_id: Option, ) -> Result, Self::Error> { - Ok(self - .negotiated_extensions - .super_safe_lock(|data| data.clone())) + self.negotiated_extensions + .get() + .map_err(TproxyError::shutdown) } async fn handle_request_extensions_success( @@ -55,9 +55,9 @@ impl HandleExtensionsFromServerAsync for ChannelManager { } // Store the negotiated extensions in the shared channel manager data - self.negotiated_extensions.super_safe_lock(|data| { - *data = supported; - }); + self.negotiated_extensions + .set(supported) + .map_err(TproxyError::shutdown)?; info!("Successfully negotiated extensions"); diff --git a/miner-apps/translator/src/lib/sv2/channel_manager/mining_message_handler.rs b/miner-apps/translator/src/lib/sv2/channel_manager/mining_message_handler.rs index b72095399..dbe5158e1 100644 --- a/miner-apps/translator/src/lib/sv2/channel_manager/mining_message_handler.rs +++ b/miner-apps/translator/src/lib/sv2/channel_manager/mining_message_handler.rs @@ -44,9 +44,9 @@ impl HandleMiningMessagesFromServerAsync for ChannelManager { &self, _server_id: Option, ) -> Result, Self::Error> { - Ok(self - .negotiated_extensions - .super_safe_lock(|data| data.clone())) + self.negotiated_extensions + .get() + .map_err(TproxyError::shutdown) } async fn handle_open_standard_mining_channel_success( @@ -88,9 +88,10 @@ impl HandleMiningMessagesFromServerAsync for ChannelManager { let full_extranonce_size = m.extranonce_size as usize + m.extranonce_prefix.len(); - // add the channel to the group channel - match self.group_channels.get_mut(&m.group_channel_id) { - Some(mut group_channel) => { + self.group_channels.with_mut_or_insert_with( + m.group_channel_id, + || GroupChannel::new(m.group_channel_id), + |group_channel| { group_channel .add_channel_id(m.channel_id, full_extranonce_size) .map_err(|e| { @@ -98,22 +99,9 @@ impl HandleMiningMessagesFromServerAsync for ChannelManager { TproxyError::fallback( TproxyErrorKind::FailedToAddChannelIdToGroupChannel(e), ) - })?; - } - None => { - let mut group_channel = GroupChannel::new(m.group_channel_id); - group_channel - .add_channel_id(m.channel_id, full_extranonce_size) - .map_err(|e| { - error!("Failed to add channel id to group channel: {:?}", e); - TproxyError::fallback( - TproxyErrorKind::FailedToAddChannelIdToGroupChannel(e), - ) - })?; - self.group_channels - .insert(m.group_channel_id, group_channel); - } - } + }) + }, + )?; let upstream_prefix_bytes = m.extranonce_prefix.to_owned_bytes(); let target = Target::from_le_bytes(m.target.to_array()); @@ -215,7 +203,8 @@ impl HandleMiningMessagesFromServerAsync for ChannelManager { // this aggregated upstream draw from the same allocator and // share rewriting reads `upstream_prefix_len()` from it. self.aggregated_extranonce_allocator - .super_safe_lock(|slot| *slot = Some(allocator)); + .set(Some(allocator)) + .map_err(TproxyError::shutdown)?; self.aggregated_channel_state .set(AggregatedState::Connected); @@ -370,18 +359,16 @@ impl HandleMiningMessagesFromServerAsync for ChannelManager { // In aggregated mode, serve any downstream requests that were buffered in // pending_channels while the upstream channel was being established (Pending state). if self.mode.is_aggregated() { - let pending_requests: Vec<(u32, String, Hashrate, usize)> = self - .pending_downstream_channels - .iter() - .map(|r| { - ( - *r.key() as u32, - r.value().0.clone(), - r.value().1, - r.value().2, - ) - }) - .collect(); + let mut pending_requests: Vec<(u32, String, Hashrate, usize)> = Vec::new(); + self.pending_downstream_channels + .for_each(|request_id, request| { + pending_requests.push(( + request_id as u32, + request.0.clone(), + request.1, + request.2, + )); + }); self.pending_downstream_channels.clear(); for (req_id, user_identity, hashrate, min_extranonce_size) in pending_requests { @@ -452,11 +439,11 @@ impl HandleMiningMessagesFromServerAsync for ChannelManager { self.extended_channels.remove(&m.channel_id); // remove the channel from any group channels that contain it - for mut group_channel in self.group_channels.iter_mut() { + self.group_channels.for_each_mut(|_, group_channel| { if group_channel.has_channel_id(m.channel_id) { group_channel.remove_channel_id(m.channel_id); } - } + }); } else { error!( "Channel Id not found: {}, ignoring CloseChannel message", @@ -488,8 +475,8 @@ impl HandleMiningMessagesFromServerAsync for ChannelManager { info!("Received: {} ✅", m); // In aggregated mode, the Pool responds with the upstream channel ID, but the - // channel is stored under AGGREGATED_CHANNEL_ID in the DashMap. - // In non-aggregated mode, m.channel_id matches the DashMap key directly. + // channel is stored under AGGREGATED_CHANNEL_ID in the shared channel map. + // In non-aggregated mode, m.channel_id matches the shared channel map key directly. let key = if self.mode.is_aggregated() { AGGREGATED_CHANNEL_ID } else { @@ -497,9 +484,9 @@ impl HandleMiningMessagesFromServerAsync for ChannelManager { }; // if None, the channel may be closed/missing, so we ignore this accounting update - if let Some(mut ch) = self.extended_channels.get_mut(&key) { + self.extended_channels.with_mut(&key, |ch| { ch.on_share_acknowledgement(m.new_submits_accepted_count, m.new_shares_sum); - } + }); Ok(()) } @@ -520,9 +507,9 @@ impl HandleMiningMessagesFromServerAsync for ChannelManager { }; // if None, the channel may be closed/missing, so we ignore this accounting update - if let Some(mut ch) = self.extended_channels.get_mut(&key) { + self.extended_channels.with_mut(&key, |ch| { ch.on_share_rejection(error_code); - } + }); Ok(()) } @@ -555,45 +542,45 @@ impl HandleMiningMessagesFromServerAsync for ChannelManager { // are we in aggregated mode? if self.mode.is_aggregated() { // Validate that the message is for the aggregated channel or its group - let (aggregated_channel_id, full_extranonce_size) = { - let aggregated_channel = self - .extended_channels - .get(&AGGREGATED_CHANNEL_ID) - .ok_or(TproxyError::fallback(TproxyErrorKind::ChannelNotFound))?; - ( - aggregated_channel.get_channel_id(), - aggregated_channel.get_full_extranonce_size(), - ) - }; + let (aggregated_channel_id, full_extranonce_size) = self + .extended_channels + .with(&AGGREGATED_CHANNEL_ID, |aggregated_channel| { + ( + aggregated_channel.get_channel_id(), + aggregated_channel.get_full_extranonce_size(), + ) + }) + .ok_or(TproxyError::fallback(TproxyErrorKind::ChannelNotFound))?; // here, we are assuming that since we are in aggregated mode, there should // be only one single group channel and the // aggregated channel must belong to it - let group_channel = self.group_channels.iter().next(); - let Some(group_channel) = group_channel else { + let mut group_channel_id = None; + self.group_channels.for_each(|channel_id, _| { + group_channel_id.get_or_insert(channel_id); + }); + let Some(group_channel_id) = group_channel_id else { error!("Aggregated channel does not belong to any group channel"); return Err(TproxyError::fallback(TproxyErrorKind::ChannelNotFound)); }; - let group_channel_id = group_channel.get_group_channel_id(); - // was the message sent to the aggregated channel? if aggregated_channel_id == m_static.channel_id || group_channel_id == m_static.channel_id { self.verify_payout_distribution(&m_static, full_extranonce_size)?; - // update all extended channel states - for mut extended_channel in self.extended_channels.iter_mut() { - extended_channel - .on_new_extended_mining_job(m_static.clone()) - .map_err(|e| { - error!("Failed to process new extended mining job: {:?}", e); - TproxyError::fallback( - TproxyErrorKind::FailedToProcessNewExtendedMiningJob, - ) - })?; - } + self.extended_channels + .try_for_each_mut(|_, extended_channel| { + extended_channel + .on_new_extended_mining_job(m_static.clone()) + .map_err(|e| { + error!("Failed to process new extended mining job: {:?}", e); + TproxyError::fallback( + TproxyErrorKind::FailedToProcessNewExtendedMiningJob, + ) + }) + })?; // only send this message to the SV1Server if it's not a future job if !m_static.is_future() { @@ -613,48 +600,90 @@ impl HandleMiningMessagesFromServerAsync for ChannelManager { } // we're not in aggregated mode // was the message sent to a group channel? - } else if let Some(mut group_channel) = self.group_channels.get_mut(&m.channel_id) { - let full_extranonce_size = - group_channel.get_full_extranonce_size().ok_or_else(|| { - error!("Group channel {} has no full extranonce size", m.channel_id); - TproxyError::fallback(TproxyErrorKind::ChannelNotFound) - })?; - self.verify_payout_distribution(&m_static, full_extranonce_size)?; - - // update group channel state - group_channel.on_new_extended_mining_job(m_static.clone()); - - // process the message for each individual channel on the group - for channel_id in group_channel.get_channel_ids() { - let mut channel = self - .extended_channels - .get_mut(channel_id) - .ok_or(TproxyError::fallback(TproxyErrorKind::ChannelNotFound))?; - - let mut job = m_static.clone(); - job.channel_id = *channel_id; - - // update each channel state - channel - .on_new_extended_mining_job(job.clone()) - .map_err(|e| { - error!("Failed to process new extended mining job: {:?}", e); - TproxyError::fallback( - TproxyErrorKind::FailedToProcessNewExtendedMiningJob, - ) - })?; - - // only send this message to the SV1Server if it's not a future job - if !job.is_future() { - new_extended_mining_job_messages.push(job); - } - } + } else if self.group_channels.contains_key(&m.channel_id) { + let messages = self + .group_channels + .with_mut(&m.channel_id, |group_channel| { + let full_extranonce_size = + group_channel.get_full_extranonce_size().ok_or_else(|| { + error!( + "Group channel {} has no full extranonce size", + m.channel_id + ); + TproxyError::fallback(TproxyErrorKind::ChannelNotFound) + })?; + self.verify_payout_distribution(&m_static, full_extranonce_size)?; + + // update group channel state + group_channel.on_new_extended_mining_job(m_static.clone()); + let channel_ids: Vec<_> = + group_channel.get_channel_ids().copied().collect(); + let mut messages = Vec::new(); + + // process the message for each individual channel on the group + for channel_id in channel_ids { + let message = self + .extended_channels + .with_mut(&channel_id, |channel| { + let mut job = m_static.clone(); + job.channel_id = channel_id; + + // update each channel state + channel.on_new_extended_mining_job(job.clone()).map_err( + |e| { + error!( + "Failed to process new extended mining job: {:?}", + e + ); + TproxyError::fallback( + TproxyErrorKind::FailedToProcessNewExtendedMiningJob, + ) + }, + )?; + + Ok::<_, Self::Error>(if !job.is_future() { + Some(job) + } else { + None + }) + }) + .ok_or(TproxyError::fallback(TproxyErrorKind::ChannelNotFound))??; + if let Some(message) = message { + messages.push(message); + } + } + Ok::<_, Self::Error>(messages) + }) + .ok_or(TproxyError::fallback(TproxyErrorKind::ChannelNotFound))??; + new_extended_mining_job_messages.extend(messages); // if the message was not sent to a group channel, we need to check if we're // working in aggregated mode } else { - let Some(mut channel) = self.extended_channels.get_mut(&m_static.channel_id) else { - // we got a nonsense channel id, we should log an error and ignore the - // message + let message = self + .extended_channels + .with_mut(&m_static.channel_id, |channel| { + self.verify_payout_distribution( + &m_static, + channel.get_full_extranonce_size(), + )?; + + // update channel state + channel + .on_new_extended_mining_job(m_static.clone()) + .map_err(|e| { + error!("Failed to process new extended mining job: {:?}", e); + TproxyError::fallback( + TproxyErrorKind::FailedToProcessNewExtendedMiningJob, + ) + })?; + + Ok::<_, Self::Error>(if !m_static.is_future() { + Some(m_static.clone()) + } else { + None + }) + }); + let Some(message) = message else { error!( "Channel not found: {}, ignoring NewExtendedMiningJob message", m_static.channel_id @@ -662,20 +691,9 @@ impl HandleMiningMessagesFromServerAsync for ChannelManager { return Err(TproxyError::log(TproxyErrorKind::ChannelNotFound)); }; - self.verify_payout_distribution(&m_static, channel.get_full_extranonce_size())?; - - // update channel state - channel - .on_new_extended_mining_job(m_static.clone()) - .map_err(|e| { - error!("Failed to process new extended mining job: {:?}", e); - TproxyError::fallback(TproxyErrorKind::FailedToProcessNewExtendedMiningJob) - })?; - // only send this message to the SV1Server if it's not a future job - if !m_static.is_future() { - let new_extended_mining_job_message = m_static.clone(); - new_extended_mining_job_messages.push(new_extended_mining_job_message); + if let Some(message) = message? { + new_extended_mining_job_messages.push(message); } } Ok::>, Self::Error>(new_extended_mining_job_messages) @@ -715,35 +733,36 @@ impl HandleMiningMessagesFromServerAsync for ChannelManager { // Validate that the message is for the aggregated channel or its group let aggregated_channel_id = self .extended_channels - .get(&AGGREGATED_CHANNEL_ID) - .ok_or(TproxyError::fallback(TproxyErrorKind::ChannelNotFound))? - .get_channel_id(); + .with(&AGGREGATED_CHANNEL_ID, |channel| channel.get_channel_id()) + .ok_or(TproxyError::fallback(TproxyErrorKind::ChannelNotFound))?; // does aggregated channel belong to some group channel? // here, we are assuming that since we are in aggregated mode, there // should be only one single group channel // and the aggregated channel must belong to it - let group_channel = self.group_channels.iter().next(); - let Some(group_channel) = group_channel else { + let mut group_channel_id = None; + self.group_channels.for_each(|channel_id, _| { + group_channel_id.get_or_insert(channel_id); + }); + let Some(group_channel_id) = group_channel_id else { error!("Aggregated channel does not belong to any group channel"); return Err(TproxyError::fallback(TproxyErrorKind::ChannelNotFound)); }; - let group_channel_id = group_channel.get_group_channel_id(); - // was the message sent to the aggregated channel? if aggregated_channel_id == m.channel_id || group_channel_id == m.channel_id { // update all extended channel states - for mut extended_channel in self.extended_channels.iter_mut() { - extended_channel - .on_set_new_prev_hash(m_static.clone()) - .map_err(|e| { - error!("Failed to set new prev hash: {:?}", e); - TproxyError::fallback( - TproxyErrorKind::FailedToProcessSetNewPrevHash, - ) - })?; - } + self.extended_channels + .try_for_each_mut(|_, extended_channel| { + extended_channel + .on_set_new_prev_hash(m_static.clone()) + .map_err(|e| { + error!("Failed to set new prev hash: {:?}", e); + TproxyError::fallback( + TproxyErrorKind::FailedToProcessSetNewPrevHash, + ) + }) + })?; // make sure the SetNewPrevHash message is sent to the aggregated // channel @@ -754,11 +773,13 @@ impl HandleMiningMessagesFromServerAsync for ChannelManager { // to the SV1Server (get active job after updating all channels) let mut new_extended_mining_job_message = self .extended_channels - .get(&AGGREGATED_CHANNEL_ID) - .expect("aggregated channel must exist") - .get_active_job() - .expect("active job must exist") - .clone(); + .with(&AGGREGATED_CHANNEL_ID, |channel| { + channel + .get_active_job() + .expect("active job must exist") + .clone() + }) + .expect("aggregated channel must exist"); new_extended_mining_job_message.0.channel_id = AGGREGATED_CHANNEL_ID; new_extended_mining_job_messages.push(new_extended_mining_job_message.0); } else { @@ -771,51 +792,88 @@ impl HandleMiningMessagesFromServerAsync for ChannelManager { return Err(TproxyError::log(TproxyErrorKind::ChannelNotFound)); } // we are not in aggregated mode.. was the message sent to a group channel? - } else if let Some(mut group_channel) = self.group_channels.get_mut(&m.channel_id) { + } else if self.group_channels.contains_key(&m.channel_id) { // update group channel state - group_channel - .on_set_new_prev_hash(m_static.clone()) - .map_err(|e| { - error!("Failed to set new prev hash: {:?}", e); - TproxyError::fallback(TproxyErrorKind::FailedToProcessSetNewPrevHash) - })?; - - // there's no aggregated channel, so we need to process the message for each - // individual channel on the group - for channel_id in group_channel.get_channel_ids() { - let mut channel = self - .extended_channels - .get_mut(channel_id) - .ok_or(TproxyError::fallback(TproxyErrorKind::ChannelNotFound))?; - - channel - .on_set_new_prev_hash(m_static.clone()) - .map_err(|e| { - error!("Failed to set new prev hash: {:?}", e); - TproxyError::fallback( - TproxyErrorKind::FailedToProcessSetNewPrevHash, - ) - })?; - - // for each extended channel, send one SetNewPrevHash message to the - // SV1Server - let mut set_new_prev_hash_message = m_static.clone(); - set_new_prev_hash_message.channel_id = *channel_id; - set_new_prev_hash_messages.push(set_new_prev_hash_message); - - // for each extended channel, send one NewExtendedMiningJob message to - // the SV1Server - let new_extended_mining_job_message = channel - .get_active_job() - .expect("active job must exist") - .clone(); - new_extended_mining_job_messages.push(new_extended_mining_job_message.0); - } + let messages = self + .group_channels + .with_mut(&m.channel_id, |group_channel| { + group_channel + .on_set_new_prev_hash(m_static.clone()) + .map_err(|e| { + error!("Failed to set new prev hash: {:?}", e); + TproxyError::fallback( + TproxyErrorKind::FailedToProcessSetNewPrevHash, + ) + })?; + let channel_ids: Vec<_> = + group_channel.get_channel_ids().copied().collect(); + let mut set_new_prev_hash_messages = Vec::new(); + let mut new_extended_mining_job_messages = Vec::new(); + + // there's no aggregated channel, so we need to process the message for + // each individual channel on the group + for channel_id in channel_ids { + let new_extended_mining_job_message = self + .extended_channels + .with_mut(&channel_id, |channel| { + channel.on_set_new_prev_hash(m_static.clone()).map_err( + |e| { + error!("Failed to set new prev hash: {:?}", e); + TproxyError::fallback( + TproxyErrorKind::FailedToProcessSetNewPrevHash, + ) + }, + )?; + + let new_extended_mining_job_message = channel + .get_active_job() + .expect("active job must exist") + .clone(); + Ok::<_, Self::Error>(new_extended_mining_job_message.0) + }) + .ok_or(TproxyError::fallback( + TproxyErrorKind::ChannelNotFound, + ))??; + + // for each extended channel, send one SetNewPrevHash message to + // the SV1Server + let mut set_new_prev_hash_message = m_static.clone(); + set_new_prev_hash_message.channel_id = channel_id; + set_new_prev_hash_messages.push(set_new_prev_hash_message); + new_extended_mining_job_messages + .push(new_extended_mining_job_message); + } + + Ok::<_, Self::Error>(( + set_new_prev_hash_messages, + new_extended_mining_job_messages, + )) + }) + .ok_or(TproxyError::fallback(TproxyErrorKind::ChannelNotFound))??; + set_new_prev_hash_messages.extend(messages.0); + new_extended_mining_job_messages.extend(messages.1); // if the message was not sent to a group channel, and we're not in aggregated // mode, we need to process the message for a specific channel } else { - let Some(mut channel) = self.extended_channels.get_mut(&m_static.channel_id) - else { + let messages = + self.extended_channels + .with_mut(&m_static.channel_id, |channel| { + channel + .on_set_new_prev_hash(m_static.clone()) + .map_err(|e| { + error!("Failed to set new prev hash: {:?}", e); + TproxyError::fallback( + TproxyErrorKind::FailedToProcessSetNewPrevHash, + ) + })?; + + let new_extended_mining_job_message = channel + .get_active_job() + .expect("active job must exist") + .clone(); + Ok::<_, Self::Error>(new_extended_mining_job_message.0) + }); + let Some(new_extended_mining_job_message) = messages else { // we got a nonsense channel id, we should log an error and ignore the // message warn!( @@ -825,23 +883,11 @@ impl HandleMiningMessagesFromServerAsync for ChannelManager { return Err(TproxyError::log(TproxyErrorKind::ChannelNotFound)); }; - // update channel state - channel - .on_set_new_prev_hash(m_static.clone()) - .map_err(|e| { - error!("Failed to set new prev hash: {:?}", e); - TproxyError::fallback(TproxyErrorKind::FailedToProcessSetNewPrevHash) - })?; - // make sure the SetNewPrevHash message is sent to the channel set_new_prev_hash_messages.push(m_static.clone()); // for the channel, send one NewExtendedMiningJob message to the SV1Server - let new_extended_mining_job_message = channel - .get_active_job() - .expect("active job must exist") - .clone(); - new_extended_mining_job_messages.push(new_extended_mining_job_message.0); + new_extended_mining_job_messages.push(new_extended_mining_job_message?); } Ok::< ( @@ -925,26 +971,26 @@ impl HandleMiningMessagesFromServerAsync for ChannelManager { if self.mode.is_aggregated() { let aggregated_channel_id = self .extended_channels - .get(&AGGREGATED_CHANNEL_ID) - .ok_or(TproxyError::fallback(TproxyErrorKind::ChannelNotFound))? - .get_channel_id(); + .with(&AGGREGATED_CHANNEL_ID, |channel| channel.get_channel_id()) + .ok_or(TproxyError::fallback(TproxyErrorKind::ChannelNotFound))?; // does aggregated channel belong to some group channel? // here, we are assuming that since we are in aggregated mode, there should // be only one single group channel and the // aggregated channel must belong to it - let group_channel = self.group_channels.iter().next(); - let Some(group_channel) = group_channel else { + let mut group_channel_id = None; + self.group_channels.for_each(|channel_id, _| { + group_channel_id.get_or_insert(channel_id); + }); + let Some(group_channel_id) = group_channel_id else { error!("Aggregated channel does not belong to any group channel"); return Err(TproxyError::fallback(TproxyErrorKind::ChannelNotFound)); }; - let group_channel_id = group_channel.get_group_channel_id(); - // was the message sent to the aggregated channel? if aggregated_channel_id == m.channel_id || group_channel_id == m.channel_id { // Update target for all extended channels (including AGGREGATED_CHANNEL_ID) - self.extended_channels.iter_mut().for_each(|mut channel| { + self.extended_channels.for_each_mut(|_, channel| { channel.set_target(Target::from_le_bytes(m.maximum_target.to_array())); }); @@ -962,24 +1008,31 @@ impl HandleMiningMessagesFromServerAsync for ChannelManager { } // we are not in aggregated mode... was the message sent to a group channel? - } else if let Some(group_channel) = self.group_channels.get(&m.channel_id) { + } else if self.group_channels.contains_key(&m.channel_id) { // process the message for each individual channel on the group - for channel_id in group_channel.get_channel_ids() { - let mut channel = self - .extended_channels - .get_mut(channel_id) + let channel_ids = self + .group_channels + .with(&m.channel_id, |group_channel| { + group_channel.get_channel_ids().copied().collect::>() + }) + .ok_or(TproxyError::fallback(TproxyErrorKind::ChannelNotFound))?; + for channel_id in channel_ids { + self.extended_channels + .with_mut(&channel_id, |channel| { + channel.set_target(Target::from_le_bytes(m.maximum_target.to_array())); + }) .ok_or(TproxyError::fallback(TproxyErrorKind::ChannelNotFound))?; - channel.set_target(Target::from_le_bytes(m.maximum_target.to_array())); - let mut message = m_static.clone(); - message.channel_id = *channel_id; + message.channel_id = channel_id; set_target_messages.push(message); } // if the message was not sent to a group channel, and we're not in aggregated // mode, we need to process the message for a specific channel } else { - let Some(mut channel) = self.extended_channels.get_mut(&m.channel_id) else { + let Some(()) = self.extended_channels.with_mut(&m.channel_id, |channel| { + channel.set_target(Target::from_le_bytes(m.maximum_target.to_array())); + }) else { // we got a nonsense channel id, we should log an error and ignore the // message warn!( @@ -989,8 +1042,6 @@ impl HandleMiningMessagesFromServerAsync for ChannelManager { return Err(TproxyError::log(TproxyErrorKind::ChannelNotFound)); }; - channel.set_target(Target::from_le_bytes(m.maximum_target.to_array())); - set_target_messages.push(m_static.clone()); } @@ -1025,32 +1076,30 @@ impl HandleMiningMessagesFromServerAsync for ChannelManager { // check every group channel if it contains any of the channels in the new group // channel - for mut channel in self.group_channels.iter_mut() { - let group_channel_id = *channel.key(); - let group_channel = channel.value_mut(); - - let channel_ids_to_remove = m.channel_ids.clone().into_inner(); - for channel_id in channel_ids_to_remove { - group_channel.remove_channel_id(channel_id); - } + self.group_channels + .for_each_mut(|group_channel_id, group_channel| { + let channel_ids_to_remove = m.channel_ids.clone().into_inner(); + for channel_id in channel_ids_to_remove { + group_channel.remove_channel_id(channel_id); + } - if group_channel.is_empty() { - group_channels_to_remove.push(group_channel_id); - } - } + if group_channel.is_empty() { + group_channels_to_remove.push(group_channel_id); + } + }); // Now remove the empty group channels for group_channel_id in group_channels_to_remove { self.group_channels.remove(&group_channel_id); } - // does the group channel already exist? - match self.group_channels.get_mut(&m.group_channel_id) { - // if yes, clean up any channels that are no longer in the new group channel - Some(mut group_channel) => { + let new_channel_ids = m.channel_ids.clone().into_inner(); + self.group_channels.with_mut_or_insert_with( + m.group_channel_id, + || GroupChannel::new(m.group_channel_id), + |group_channel| { let current_channel_ids: Vec = group_channel.get_channel_ids().copied().collect(); - let new_channel_ids = m.channel_ids.clone().into_inner(); // Remove channels that are no longer in the new list for channel_id in ¤t_channel_ids { @@ -1061,12 +1110,12 @@ impl HandleMiningMessagesFromServerAsync for ChannelManager { // Add all channels from the message (inner HashSet ingores duplicates) for channel_id in new_channel_ids { - let extended_channel = self + let full_extranonce_size = self .extended_channels - .get(&channel_id) + .with(&channel_id, |extended_channel| { + extended_channel.get_full_extranonce_size() + }) .ok_or_else(|| TproxyError::fallback(TproxyErrorKind::ChannelNotFound))?; - - let full_extranonce_size = extended_channel.get_full_extranonce_size(); group_channel .add_channel_id(channel_id, full_extranonce_size) .map_err(|e| { @@ -1076,34 +1125,9 @@ impl HandleMiningMessagesFromServerAsync for ChannelManager { ) })?; } - } - // if no, create a new group channel, and add all the channels to it - None => { - let mut group_channel = GroupChannel::new(m.group_channel_id); - - // Add all channels to the newly created group channel - for channel_id in m.channel_ids.clone().into_inner() { - let extended_channel = self - .extended_channels - .get(&channel_id) - .ok_or_else(|| TproxyError::fallback(TproxyErrorKind::ChannelNotFound))?; - - let full_extranonce_size = extended_channel.get_full_extranonce_size(); - - group_channel - .add_channel_id(channel_id, full_extranonce_size) - .map_err(|e| { - error!("Failed to add channel id to group channel: {:?}", e); - TproxyError::fallback( - TproxyErrorKind::FailedToAddChannelIdToGroupChannel(e), - ) - })?; - } - - self.group_channels - .insert(m.group_channel_id, group_channel); - } - } + Ok::<(), Self::Error>(()) + }, + )?; Ok(()) } diff --git a/miner-apps/translator/src/lib/sv2/channel_manager/mod.rs b/miner-apps/translator/src/lib/sv2/channel_manager/mod.rs index 830d87250..047bf0ba9 100644 --- a/miner-apps/translator/src/lib/sv2/channel_manager/mod.rs +++ b/miner-apps/translator/src/lib/sv2/channel_manager/mod.rs @@ -10,11 +10,9 @@ use crate::{ TproxyMode, }; use async_channel::{Receiver, Sender}; -use dashmap::DashMap; use std::sync::{Arc, OnceLock}; use stratum_apps::{ channel_utils::ReceiverCleanup, - custom_mutex::Mutex, fallback_coordinator::FallbackCoordinator, payout::PayoutMode, stratum_core::{ @@ -29,6 +27,7 @@ use stratum_apps::{ mining_sv2::OpenExtendedMiningChannelSuccess, parsers_sv2::{AnyMessage, Mining, TlvField, TlvList}, }, + sync::{SharedLock, SharedMap}, task_manager::TaskManager, utils::{ protocol_message_type::{protocol_message_type, MessageType}, @@ -147,25 +146,25 @@ pub struct ChannelManager { /// /// Entries are removed once the upstream success message is received /// and propagated accordingly. - pub pending_downstream_channels: Arc>, + pub pending_downstream_channels: SharedMap, /// Map of active extended channels by channel ID. /// In aggregated mode, the shared upstream channel is stored under AGGREGATED_CHANNEL_ID. /// In non-aggregated mode, each downstream has its own channel with its assigned ID. - pub extended_channels: Arc>>, + pub extended_channels: SharedMap>, /// Map of active group channels by group channel ID - pub group_channels: Arc>>, + pub group_channels: SharedMap>, /// Share sequence number counter for tracking valid shares forwarded upstream. /// In aggregated mode: single counter for all shares going to the upstream channel. /// In non-aggregated mode: one counter per downstream channel. - pub share_sequence_counters: Arc>, + pub share_sequence_counters: SharedMap, /// Extensions that have been successfully negotiated with the upstream server - pub negotiated_extensions: Arc>>, + pub negotiated_extensions: SharedLock>, /// Single extranonce allocator used in aggregated mode to sub-divide the /// upstream-assigned prefix across all downstream channels. /// /// `None` until the upstream `OpenExtendedMiningChannelSuccess` for the /// aggregated channel is received; `Some` afterwards. - pub aggregated_extranonce_allocator: Arc>>, + pub aggregated_extranonce_allocator: SharedLock>, /// Tracks whether the single upstream channel in aggregated mode is absent, /// being established, or connected. pub aggregated_channel_state: AtomicAggregatedState, @@ -287,12 +286,12 @@ impl ChannelManager { channel_manager_io, supported_extensions, required_extensions, - pending_downstream_channels: Arc::new(DashMap::new()), - extended_channels: Arc::new(DashMap::new()), - group_channels: Arc::new(DashMap::new()), - share_sequence_counters: Arc::new(DashMap::new()), - negotiated_extensions: Arc::new(Mutex::new(Vec::new())), - aggregated_extranonce_allocator: Arc::new(Mutex::new(None)), + pending_downstream_channels: SharedMap::new(), + extended_channels: SharedMap::new(), + group_channels: SharedMap::new(), + share_sequence_counters: SharedMap::new(), + negotiated_extensions: SharedLock::new(Vec::new()), + aggregated_extranonce_allocator: SharedLock::new(None), aggregated_channel_state: AtomicAggregatedState::new(AggregatedState::NoChannel), expected_payout_distribution: Arc::new(OnceLock::new()), mode: tproxy_mode, @@ -553,13 +552,15 @@ impl ChannelManager { let downstream_channel_id = m.channel_id; let downstream_extranonce_prefix = self .extended_channels - .get(&downstream_channel_id) - .map(|channel| channel.get_extranonce_prefix().to_vec()); - let upstream_prefix_len = - self.aggregated_extranonce_allocator - .super_safe_lock(|allocator| { - allocator.as_ref().map(|a| a.upstream_prefix_len() as usize) - }); + .with(&downstream_channel_id, |channel| { + channel.get_extranonce_prefix().to_vec() + }); + let upstream_prefix_len = self + .aggregated_extranonce_allocator + .with(|allocator| { + allocator.as_ref().map(|a| a.upstream_prefix_len() as usize) + }) + .map_err(TproxyError::shutdown)?; if let (Some(downstream_extranonce_prefix), Some(upstream_prefix_len)) = (downstream_extranonce_prefix, upstream_prefix_len) { @@ -575,15 +576,15 @@ impl ChannelManager { let upstream_extended_channel_id = self .extended_channels - .get(&AGGREGATED_CHANNEL_ID) - .map(|ch| ch.get_channel_id()) + .with(&AGGREGATED_CHANNEL_ID, |ch| ch.get_channel_id()) .unwrap(); m.channel_id = upstream_extended_channel_id; let value = self .extended_channels - .get_mut(&AGGREGATED_CHANNEL_ID) - .map(|mut aggregated_channel| aggregated_channel.validate_share(m.clone())); + .with_mut(&AGGREGATED_CHANNEL_ID, |aggregated_channel| { + aggregated_channel.validate_share(m.clone()) + }); if let Some(Ok(_result)) = value { info!( "SubmitSharesExtended: valid share, forwarding it to upstream | channel_id: {}, sequence_number: {} ☑️", @@ -599,8 +600,9 @@ impl ChannelManager { } else { let value = self .extended_channels - .get_mut(&m.channel_id) - .map(|mut extended_channel| extended_channel.validate_share(m.clone())); + .with_mut(&m.channel_id, |extended_channel| { + extended_channel.validate_share(m.clone()) + }); if let Some(Ok(_result)) = value { info!( "SubmitSharesExtended: valid share, forwarding it to upstream | channel_id: {}, sequence_number: {} ☑️", @@ -624,10 +626,13 @@ impl ChannelManager { // zero-slack open path where upstream granted // exactly the rollable size we asked for and no // rewriting is needed. - let layout = self.extended_channels.get(&m.channel_id).and_then(|c| { - c.upstream_prefix_len() - .map(|n| (n as usize, c.get_extranonce_prefix().to_vec())) - }); + let layout = self + .extended_channels + .with(&m.channel_id, |c| { + c.upstream_prefix_len() + .map(|n| (n as usize, c.get_extranonce_prefix().to_vec())) + }) + .flatten(); if let Some((upstream_prefix_len, downstream_extranonce_prefix)) = layout { let translator_prefix = &downstream_extranonce_prefix[upstream_prefix_len..]; @@ -642,10 +647,10 @@ impl ChannelManager { } // Send the share upstream (common for both aggregated and non-aggregated modes) - let contains_type_in_negotiated_extension = - self.negotiated_extensions.super_safe_lock(|data| { - data.contains(&EXTENSION_TYPE_WORKER_HASHRATE_TRACKING) - }); + let contains_type_in_negotiated_extension = self + .negotiated_extensions + .with(|data| data.contains(&EXTENSION_TYPE_WORKER_HASHRATE_TRACKING)) + .map_err(TproxyError::shutdown)?; let mut sent = false; if contains_type_in_negotiated_extension { @@ -717,11 +722,14 @@ impl ChannelManager { // Update the aggregated channel's nominal hashrate so // that monitoring reports a value consistent with the // downstream vardiff estimate. - if let Some(mut aggregated_extended_channel) = - self.extended_channels.get_mut(&AGGREGATED_CHANNEL_ID) - { - aggregated_extended_channel.set_nominal_hashrate(m.nominal_hash_rate); - m.channel_id = aggregated_extended_channel.get_channel_id(); + if let Some(channel_id) = self.extended_channels.with_mut( + &AGGREGATED_CHANNEL_ID, + |aggregated_extended_channel| { + aggregated_extended_channel.set_nominal_hashrate(m.nominal_hash_rate); + aggregated_extended_channel.get_channel_id() + }, + ) { + m.channel_id = channel_id; } else { warn!( "Ignoring aggregated UpdateChannel before upstream channel is open: {}", @@ -731,9 +739,9 @@ impl ChannelManager { } } else { // Non-aggregated: update the specific channel's nominal hashrate - if let Some(mut channel) = self.extended_channels.get_mut(&m.channel_id) { + self.extended_channels.with_mut(&m.channel_id, |channel| { channel.set_nominal_hashrate(m.nominal_hash_rate); - } + }); } info!( @@ -788,12 +796,12 @@ impl ChannelManager { } // Remove from any group channels that contain it - for mut group_channel in self.group_channels.iter_mut() { + self.group_channels.for_each_mut(|_, group_channel| { if group_channel.has_channel_id(m.channel_id) { group_channel.remove_channel_id(m.channel_id); debug!("Removed channel {} from group channel", m.channel_id); } - } + }); // Only forward `CloseChannel` upstream in non-aggregated // mode. In aggregated mode the upstream channel is shared @@ -845,8 +853,7 @@ impl ChannelManager { // server. let target = self .extended_channels - .get(&AGGREGATED_CHANNEL_ID) - .map(|ch| *ch.get_target()) + .with(&AGGREGATED_CHANNEL_ID, |ch| *ch.get_target()) .unwrap(); // The aggregated allocator was built with the upstream prefix padded // so that `rollable_extranonce_size == config.downstream_extranonce2_size`. @@ -855,23 +862,25 @@ impl ChannelManager { // the downstream sees as its SV1 extranonce1. let allocation = self .aggregated_extranonce_allocator - .super_safe_lock(|allocator| { + .with(|allocator| { allocator.as_mut().and_then(|a| { let rollable = a.rollable_extranonce_size() as usize; a.allocate_extended(min_extranonce_size) .ok() .map(|prefix| (prefix, rollable)) }) - }); + }) + .map_err(TproxyError::shutdown)?; if let Some((new_extranonce_prefix, rollable_extranonce_size)) = allocation { if rollable_extranonce_size == min_extranonce_size { // Find max channel ID, excluding AGGREGATED_CHANNEL_ID // (u32::MAX) which would cause overflow when adding 1 - let channel_id = self - .extended_channels - .iter() - .filter(|x| *x.key() != AGGREGATED_CHANNEL_ID) - .fold(0, |acc, x| std::cmp::max(acc, *x.key())); + let mut channel_id = 0; + self.extended_channels.for_each(|extended_channel_id, _| { + if extended_channel_id != AGGREGATED_CHANNEL_ID { + channel_id = channel_id.max(extended_channel_id); + } + }); let next_channel_id = channel_id + 1; let success_extranonce_prefix: Vec = new_extranonce_prefix.as_bytes().to_vec(); let new_downstream_extended_channel = ExtendedChannel::new( @@ -912,40 +921,43 @@ impl ChannelManager { let active_job_for_sv1_server = || { // Extract data from aggregated channel in a scope block // to release the borrow before accessing other channels - let (last_active_job, future_jobs, last_chain_tip) = { - let aggregated_channel = - self.extended_channels.get(&AGGREGATED_CHANNEL_ID)?; - ( - aggregated_channel.get_active_job().map(|j| j.0.clone()), - aggregated_channel - .get_future_jobs() - .map(|(_, job)| job) - .map(|j| j.0.clone()) - .collect::>(), - aggregated_channel.get_chain_tip().cloned(), - ) - }; + let (last_active_job, future_jobs, last_chain_tip) = self + .extended_channels + .with(&AGGREGATED_CHANNEL_ID, |aggregated_channel| { + ( + aggregated_channel.get_active_job().map(|j| j.0.clone()), + aggregated_channel + .get_future_jobs() + .map(|(_, job)| job) + .map(|j| j.0.clone()) + .collect::>(), + aggregated_channel.get_chain_tip().cloned(), + ) + })?; if let Some(chain_tip) = last_chain_tip { self.extended_channels - .get_mut(&next_channel_id)? - .set_chain_tip(chain_tip); + .with_mut(&next_channel_id, |channel| { + channel.set_chain_tip(chain_tip) + })?; } if let Some(mut job) = last_active_job.clone() { job.channel_id = next_channel_id; _ = self .extended_channels - .get_mut(&next_channel_id)? - .on_new_extended_mining_job(job); + .with_mut(&next_channel_id, |channel| { + channel.on_new_extended_mining_job(job) + })?; } // Also add any future jobs so SetNewPrevHash won't fail for mut future_job in future_jobs { future_job.channel_id = next_channel_id; _ = self .extended_channels - .get_mut(&next_channel_id)? - .on_new_extended_mining_job(future_job); + .with_mut(&next_channel_id, |channel| { + channel.on_new_extended_mining_job(future_job) + })?; } last_active_job.map(|mut job| { @@ -978,11 +990,11 @@ impl ChannelManager { /// - In aggregated mode: use upstream channel ID (single counter for all shares) /// - In non-aggregated mode: use downstream channel ID (one counter per channel) fn next_share_sequence_number(&self, counter_key: u32) -> u32 { - let mut counter = self.share_sequence_counters.entry(counter_key).or_insert(0); - let counter = counter.value_mut(); - - *counter += 1; - *counter + self.share_sequence_counters + .with_mut_or_default(counter_key, |counter| { + *counter += 1; + *counter + }) } } From a03491cc94f2da0f23a93d0fd6657eecf6b61bfe Mon Sep 17 00:00:00 2001 From: bit-aloo Date: Tue, 21 Jul 2026 17:44:14 +0530 Subject: [PATCH 2/3] Remove dashmap from translator --- integration-tests/Cargo.lock | 1 - miner-apps/Cargo.lock | 1 - miner-apps/translator/Cargo.toml | 1 - .../translator/src/lib/sv1/sv1_server/difficulty_manager.rs | 1 + 4 files changed, 1 insertion(+), 3 deletions(-) diff --git a/integration-tests/Cargo.lock b/integration-tests/Cargo.lock index 9b309668e..102fddbc7 100644 --- a/integration-tests/Cargo.lock +++ b/integration-tests/Cargo.lock @@ -4901,7 +4901,6 @@ dependencies = [ "async-channel 1.9.0", "clap", "config", - "dashmap", "hex", "hotpath", "serde", diff --git a/miner-apps/Cargo.lock b/miner-apps/Cargo.lock index 62b39946e..66b872cea 100644 --- a/miner-apps/Cargo.lock +++ b/miner-apps/Cargo.lock @@ -4294,7 +4294,6 @@ dependencies = [ "async-channel", "clap", "config", - "dashmap", "hex", "hotpath", "serde", diff --git a/miner-apps/translator/Cargo.toml b/miner-apps/translator/Cargo.toml index 56dc48beb..812b4d96c 100644 --- a/miner-apps/translator/Cargo.toml +++ b/miner-apps/translator/Cargo.toml @@ -31,7 +31,6 @@ tracing = { version = "0.1" } clap = { version = "4.5.39", features = ["derive"] } hex = "0.4.3" hotpath = "0.14.0" -dashmap = "6.1.0" [features] default = ["monitoring"] diff --git a/miner-apps/translator/src/lib/sv1/sv1_server/difficulty_manager.rs b/miner-apps/translator/src/lib/sv1/sv1_server/difficulty_manager.rs index cc15bbcdf..18ee00a9e 100644 --- a/miner-apps/translator/src/lib/sv1/sv1_server/difficulty_manager.rs +++ b/miner-apps/translator/src/lib/sv1/sv1_server/difficulty_manager.rs @@ -467,6 +467,7 @@ impl Sv1Server { /// Gets pending updates that can now be applied based on the new upstream target. /// If downstream_id is provided, only returns updates for that specific downstream. /// Logs a warning if the upstream target is higher than any requested target. + #[allow(clippy::result_large_err)] fn get_pending_difficulty_updates( &self, new_upstream_target: Target, From 9e1e153201e0c17d869dda99f1f0cfe733dbdd55 Mon Sep 17 00:00:00 2001 From: bit-aloo Date: Tue, 21 Jul 2026 18:52:51 +0530 Subject: [PATCH 3/3] open-channel success can orphan an upstream/SV2 channel if the SV1 downstream disconnects mid-open --- .../translator/src/lib/sv1/sv1_server/mod.rs | 32 +++++++++++++++---- 1 file changed, 26 insertions(+), 6 deletions(-) diff --git a/miner-apps/translator/src/lib/sv1/sv1_server/mod.rs b/miner-apps/translator/src/lib/sv1/sv1_server/mod.rs index 0617911fd..208b7bdc5 100644 --- a/miner-apps/translator/src/lib/sv1/sv1_server/mod.rs +++ b/miner-apps/translator/src/lib/sv1/sv1_server/mod.rs @@ -792,10 +792,6 @@ impl Sv1Server { downstream_id ); - let request_id = self.request_id_factory.fetch_add(1, Ordering::Relaxed); - self.request_id_to_downstream_id - .insert(request_id, downstream_id); - if !self.downstreams.contains_key(&downstream_id) { error!( "Downstream {} not found when attempting to open channel", @@ -807,8 +803,17 @@ impl Sv1Server { )); } - self.open_extended_mining_channel(request_id, downstream_id) - .await?; + let request_id = self.request_id_factory.fetch_add(1, Ordering::Relaxed); + self.request_id_to_downstream_id + .insert(request_id, downstream_id); + + if let Err(e) = self + .open_extended_mining_channel(request_id, downstream_id) + .await + { + self.request_id_to_downstream_id.remove(&request_id); + return Err(e); + } Ok(()) } @@ -972,6 +977,21 @@ impl Sv1Server { Err(e) => { if matches!(e.kind, TproxyErrorKind::DownstreamNotFound(_)) { error!("Downstream not found for downstream_id: {}", downstream_id); + let reason_code = + Str0255::try_from("downstream disconnected".to_string()).unwrap(); + self.sv1_server_io + .channel_manager_sender + .send(( + Mining::CloseChannel(CloseChannel { + channel_id: m.channel_id, + reason_code, + }), + None, + )) + .await + .map_err(|_| { + TproxyError::shutdown(TproxyErrorKind::ChannelErrorSender) + })?; } else { return Err(e); }