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 e16ff489c..d4747dc79 100644 --- a/miner-apps/translator/src/lib/sv1/sv1_server/mod.rs +++ b/miner-apps/translator/src/lib/sv1/sv1_server/mod.rs @@ -301,7 +301,7 @@ impl Sv1Server { } } } else { - // Non-aggregated: send to the single downstream that owns this channel_id. + // A concrete channel ID targets the single downstream that owns it. let downstream_id = match self .channel_id_to_downstream_id .with(&channel_id, |downstream_id| *downstream_id) @@ -1081,15 +1081,38 @@ impl Sv1Server { } } + MiningOwned::OpenMiningChannelError(m) => { + warn!( + request_id = m.request_id, + error_code = %m.error_code.as_utf8_or_hex(), + "Channel manager rejected downstream channel request" + ); + let downstream_id = self.request_id_to_downstream_id.remove(&m.request_id); + let Some((_, downstream_id)) = downstream_id else { + return Err(TproxyError::log(TproxyErrorKind::RequestIdNotFound( + m.request_id, + ))); + }; + return Err(TproxyError::disconnect( + TproxyErrorKind::OpenMiningChannelError, + downstream_id, + )); + } + MiningOwned::NewExtendedMiningJob(m) => { debug!( "Received NewExtendedMiningJob for channel id: {}", m.channel_id ); + let prevhash_channel_id = if self.mode.is_aggregated() { + AGGREGATED_CHANNEL_ID + } else { + m.channel_id + }; // Clone the prevhash immediately so shared map access is not held across .await. if let Some(prevhash) = self .prevhashes - .with(&m.channel_id, |prevhash| prevhash.clone()) + .with(&prevhash_channel_id, |prevhash| prevhash.clone()) { let clean_jobs = m.job_id == prevhash.job_id; let notify = build_sv1_notify_from_sv2(prevhash, m.clone(), clean_jobs) @@ -1097,23 +1120,28 @@ impl Sv1Server { // Update job storage based on the configured mode let notify_parsed = notify.clone(); - let job_channel_id = if self.mode.is_non_aggregated() { + let job_storage_channel_id = if self.mode.is_non_aggregated() { m.channel_id } else { AGGREGATED_CHANNEL_ID }; - self.valid_sv1_jobs - .with_mut_or_default(job_channel_id, |channel_jobs| { + self.valid_sv1_jobs.with_mut_or_default( + job_storage_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(); - self.send_to_channel(job_channel_id, notify_msg).await; + // Normal aggregated jobs carry AGGREGATED_CHANNEL_ID and are broadcast. A + // bootstrap job for a late joiner carries that downstream's channel ID and + // must only be delivered to that miner. + self.send_to_channel(m.channel_id, notify_msg).await; } } @@ -1750,7 +1778,13 @@ mod tests { use std::str::FromStr; use stratum_apps::{ key_utils::Secp256k1PublicKey, - stratum_core::mining_sv2::{OpenExtendedMiningChannelSuccessOwned, SetTargetOwned}, + stratum_core::{ + binary_sv2::{Seq0255Owned, Sv2OptionOwned}, + mining_sv2::{ + NewExtendedMiningJobOwned, OpenExtendedMiningChannelSuccessOwned, + OpenMiningChannelErrorOwned, SetNewPrevHashOwned, SetTargetOwned, + }, + }, }; fn create_test_config() -> TranslatorConfig { @@ -1955,6 +1989,59 @@ mod tests { assert!(server.request_id_to_downstream_id.is_empty()); } + #[tokio::test] + async fn rejected_open_request_disconnects_pending_downstream() { + let (server_to_channel_manager_sender, _server_to_channel_manager_receiver) = unbounded(); + let (channel_manager_to_server_sender, channel_manager_to_server_receiver) = unbounded(); + let config = create_test_config(); + let addr = "127.0.0.1:3333".parse().unwrap(); + let mode = TproxyMode::from(config.aggregate_channels); + let server = Sv1Server::new( + addr, + channel_manager_to_server_receiver, + server_to_channel_manager_sender, + config, + mode, + ); + register_test_downstream(&server, 7, None, 100.0, false); + server.request_id_to_downstream_id.insert(42, 7); + channel_manager_to_server_sender + .send(MiningOwned::OpenMiningChannelError( + OpenMiningChannelErrorOwned { + request_id: 42, + error_code: "channel-capacity-exhausted".try_into().unwrap(), + }, + )) + .await + .unwrap(); + + let error = server + .handle_upstream_message(Target::from_le_bytes([0xff; 32])) + .await + .unwrap_err(); + assert!(matches!(error.action, Action::Disconnect(7))); + assert!(server.request_id_to_downstream_id.is_empty()); + + let cancellation_token = CancellationToken::new(); + let fallback_token = CancellationToken::new(); + let control = server + .handle_error_action( + "rejected open request", + &error, + &cancellation_token, + &fallback_token, + ) + .await; + assert!(matches!(control, LoopControl::Continue)); + assert!(!server.downstreams.contains_key(&7)); + assert!( + !server + .sv1_server_io + .sv1_server_to_downstream_sender + .contains_key(&7) + ); + } + #[tokio::test] async fn late_open_success_closes_channel_for_disconnected_downstream() { let (server_to_channel_manager_sender, server_to_channel_manager_receiver) = unbounded(); @@ -2105,6 +2192,78 @@ mod tests { assert_eq!(update.nominal_hash_rate, 200.0); } + #[tokio::test] + async fn aggregated_targeted_job_is_not_broadcast() { + let (server_to_channel_manager_sender, _server_to_channel_manager_receiver) = unbounded(); + let (channel_manager_to_server_sender, channel_manager_to_server_receiver) = unbounded(); + let config = create_test_config(); + let addr = "127.0.0.1:3333".parse().unwrap(); + let mode = TproxyMode::from(config.aggregate_channels); + let server = Sv1Server::new( + addr, + channel_manager_to_server_receiver, + server_to_channel_manager_sender, + config, + mode, + ); + let (first_downstream_sender, first_downstream_receiver) = unbounded(); + let (second_downstream_sender, second_downstream_receiver) = unbounded(); + server + .sv1_server_io + .sv1_server_to_downstream_sender + .insert(1, first_downstream_sender); + server + .sv1_server_io + .sv1_server_to_downstream_sender + .insert(2, second_downstream_sender); + server.channel_id_to_downstream_id.insert(7, 1); + server.channel_id_to_downstream_id.insert(8, 2); + + channel_manager_to_server_sender + .send(MiningOwned::SetNewPrevHash(SetNewPrevHashOwned { + channel_id: AGGREGATED_CHANNEL_ID, + job_id: 1, + prev_hash: vec![0; 32].try_into().unwrap(), + min_ntime: 0, + nbits: 0x207fffff, + })) + .await + .unwrap(); + server + .handle_upstream_message(Target::from_le_bytes([0xff; 32])) + .await + .unwrap(); + + channel_manager_to_server_sender + .send(MiningOwned::NewExtendedMiningJob( + NewExtendedMiningJobOwned { + channel_id: 7, + job_id: 1, + min_ntime: Sv2OptionOwned::new(None), + version: 0x20000000, + version_rolling_allowed: true, + merkle_path: Seq0255Owned::new(vec![]).unwrap(), + coinbase_tx_prefix: hex::decode("02000000010000000000000000000000000000000000000000000000000000000000000000ffffffff265200162f5374726174756d2056322053524920506f6f6c2f2f08") + .unwrap() + .try_into() + .unwrap(), + coinbase_tx_suffix: hex::decode("feffffff0200f2052a01000000160014ebe1b7dcc293ccaa0ee743a86f89df8258c208fc0000000000000000266a24aa21a9ede2f61c3f71d1defd3fa999dfa36953755c690689799962b48bebd836974e8cf901000000") + .unwrap() + .try_into() + .unwrap(), + }, + )) + .await + .unwrap(); + server + .handle_upstream_message(Target::from_le_bytes([0xff; 32])) + .await + .unwrap(); + + assert!(first_downstream_receiver.try_recv().is_ok()); + assert!(second_downstream_receiver.try_recv().is_err()); + } + #[tokio::test] async fn closed_downstream_does_not_shutdown_on_aggregated_set_target() { let mut config = create_test_config(); 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 19f1eeb8b..6adae9fa3 100644 --- a/miner-apps/translator/src/lib/sv2/channel_manager/mod.rs +++ b/miner-apps/translator/src/lib/sv2/channel_manager/mod.rs @@ -26,7 +26,7 @@ use stratum_apps::{ handlers_sv2::{ HandleExtensionsFromServerOwnedAsync, HandleMiningMessagesFromServerOwnedAsync, }, - mining_sv2::OpenExtendedMiningChannelSuccessOwned, + mining_sv2::{OpenExtendedMiningChannelSuccessOwned, OpenMiningChannelErrorOwned}, parsers_sv2::{AnyMessageOwned, MiningOwned, TlvField, TlvList}, }, sync::{SharedLock, SharedMap}, @@ -69,6 +69,9 @@ pub(crate) const AGGREGATED_TPROXY_LOCAL_PREFIX_BYTES: u8 = /// and share rewriting is a no-op. pub(crate) const NON_AGGREGATED_TPROXY_MAX_CHANNELS: u32 = 1; +const ERROR_CODE_CHANNEL_CAPACITY_EXHAUSTED: &str = "channel-capacity-exhausted"; +const ERROR_CODE_INVALID_EXTRANONCE_SIZE: &str = "invalid-extranonce-size"; + #[derive(Clone, Debug)] struct ChannelManagerIo { upstream_sender: Sender, @@ -181,6 +184,32 @@ pub struct ChannelManager { #[cfg_attr(not(test), hotpath::measure_all)] impl ChannelManager { + async fn reject_downstream_channel_request( + &self, + request_id: RequestId, + error_code: &'static str, + ) -> TproxyResult<(), error::ChannelManager> { + warn!( + request_id, + error_code, "Rejecting downstream channel request" + ); + self.channel_manager_io + .sv1_server_sender + .send(MiningOwned::OpenMiningChannelError( + OpenMiningChannelErrorOwned { + request_id, + error_code: error_code + .try_into() + .expect("static channel error code must fit in Str0255"), + }, + )) + .await + .map_err(|e| { + error!("Failed to send open channel error to SV1Server: {e:?}"); + TproxyError::shutdown(TproxyErrorKind::ChannelErrorSender) + }) + } + fn expected_payout_distribution(&self) -> &Option { self.expected_payout_distribution .get() @@ -867,15 +896,19 @@ impl ChannelManager { let allocation = self .aggregated_extranonce_allocator .with(|allocator| { - allocator.as_mut().and_then(|a| { + allocator.as_mut().map(|a| { let rollable = a.rollable_extranonce_size() as usize; - a.allocate_extended(min_extranonce_size) - .ok() - .map(|prefix| (prefix, rollable)) + (a.allocate_extended(min_extranonce_size), rollable) }) }) .map_err(TproxyError::shutdown)?; - if let Some((new_extranonce_prefix, rollable_extranonce_size)) = allocation { + let Some((allocation, rollable_extranonce_size)) = allocation else { + error!("Aggregated channel is connected without an extranonce allocator"); + return Err(TproxyError::shutdown( + TproxyErrorKind::OpenMiningChannelError, + )); + }; + if let Ok(new_extranonce_prefix) = 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 @@ -885,7 +918,22 @@ impl ChannelManager { channel_id = channel_id.max(extended_channel_id); } }); - let next_channel_id = channel_id + 1; + let next_channel_id = channel_id + .checked_add(1) + .filter(|channel_id| *channel_id != AGGREGATED_CHANNEL_ID) + .or_else(|| { + (1..AGGREGATED_CHANNEL_ID) + .find(|channel_id| !self.extended_channels.contains_key(channel_id)) + }); + let Some(next_channel_id) = next_channel_id else { + drop(new_extranonce_prefix); + return self + .reject_downstream_channel_request( + request_id, + ERROR_CODE_CHANNEL_CAPACITY_EXHAUSTED, + ) + .await; + }; let success_extranonce_prefix: Vec = new_extranonce_prefix.as_bytes().to_vec(); let new_downstream_extended_channel = ExtendedChannel::new( next_channel_id, @@ -966,7 +1014,7 @@ impl ChannelManager { } last_active_job.map(|mut job| { - job.channel_id = AGGREGATED_CHANNEL_ID; + job.channel_id = next_channel_id; job }) }; @@ -984,9 +1032,19 @@ impl ChannelManager { TproxyError::shutdown(TproxyErrorKind::ChannelErrorSender) })?; } + return Ok(()); } } - Ok(()) + if rollable_extranonce_size != min_extranonce_size { + self.reject_downstream_channel_request(request_id, ERROR_CODE_INVALID_EXTRANONCE_SIZE) + .await + } else { + self.reject_downstream_channel_request( + request_id, + ERROR_CODE_CHANNEL_CAPACITY_EXHAUSTED, + ) + .await + } } /// Gets the next sequence number for a valid share and increments the counter. @@ -1007,8 +1065,14 @@ impl ChannelManager { mod tests { use super::*; use async_channel::unbounded; - use stratum_apps::stratum_core::mining_sv2::{ - OpenExtendedMiningChannelOwned, SubmitSharesExtendedOwned, UpdateChannelOwned, + use stratum_apps::stratum_core::{ + binary_sv2::{Seq0255Owned, Sv2OptionOwned}, + bitcoin::Target, + channels_sv2::extranonce_manager::ExtranoncePrefix, + mining_sv2::{ + NewExtendedMiningJobOwned, OpenExtendedMiningChannelOwned, SetNewPrevHashOwned, + SubmitSharesExtendedOwned, UpdateChannelOwned, + }, }; fn create_test_channel_manager() -> ChannelManager { @@ -1030,6 +1094,55 @@ mod tests { ) } + fn create_connected_aggregated_channel_manager() -> (ChannelManager, Receiver) { + let (upstream_sender, _upstream_receiver) = unbounded(); + let (_upstream_sender, upstream_receiver) = unbounded(); + let (sv1_server_sender, sv1_server_receiver_for_test) = unbounded(); + let (_sv1_server_sender, sv1_server_receiver) = unbounded(); + + let manager = ChannelManager::new( + upstream_sender, + upstream_receiver, + sv1_server_sender, + sv1_server_receiver, + vec![], + vec![], + TproxyMode::Aggregated, + #[cfg(feature = "monitoring")] + true, + ); + + manager.extended_channels.insert( + AGGREGATED_CHANNEL_ID, + ExtendedChannel::new( + 42, + "aggregated".to_string(), + ExtranoncePrefix::from_wire(vec![0; 4]).unwrap(), + Target::from_le_bytes([0xff; 32]), + 1.0, + true, + 8, + ), + ); + manager + .aggregated_extranonce_allocator + .set(Some( + ExtranonceAllocator::from_upstream_prefix( + vec![0; 4], + vec![], + 12, + AGGREGATED_TPROXY_MAX_CHANNELS, + ) + .unwrap(), + )) + .unwrap(); + manager + .aggregated_channel_state + .set(AggregatedState::Connected); + + (manager, sv1_server_receiver_for_test) + } + #[tokio::test] async fn test_handle_downstream_open_channel_message() { let manager = create_test_channel_manager(); @@ -1174,4 +1287,139 @@ mod tests { assert!(has_pending); } + + #[tokio::test] + async fn aggregated_channel_capacity_exhaustion_rejects_request() { + let (manager, sv1_server_receiver) = create_connected_aggregated_channel_manager(); + let mut allocator = + ExtranonceAllocator::from_upstream_prefix(vec![0; 4], vec![0; 1], 12, 1).unwrap(); + let occupied_prefix = allocator.allocate_extended(6).unwrap(); + manager + .aggregated_extranonce_allocator + .set(Some(allocator)) + .unwrap(); + + manager + .handle_downstream_channel_request_in_aggregated_mode( + 7, + "rejected-miner".to_string(), + 1.0, + 6, + ) + .await + .unwrap(); + + let error = match sv1_server_receiver.try_recv().unwrap() { + MiningOwned::OpenMiningChannelError(error) => error, + other => panic!("expected open error, got {other:?}"), + }; + assert_eq!(error.request_id, 7); + assert_eq!( + error.error_code.as_utf8_or_hex(), + ERROR_CODE_CHANNEL_CAPACITY_EXHAUSTED + ); + assert!(sv1_server_receiver.try_recv().is_err()); + + drop(occupied_prefix); + } + + #[tokio::test] + async fn aggregated_channel_id_allocation_does_not_use_reserved_id() { + let (manager, sv1_server_receiver) = create_connected_aggregated_channel_manager(); + manager.extended_channels.insert( + AGGREGATED_CHANNEL_ID - 1, + ExtendedChannel::new( + AGGREGATED_CHANNEL_ID - 1, + "last-channel".to_string(), + ExtranoncePrefix::from_wire(vec![0xaa; 6]).unwrap(), + Target::from_le_bytes([0xff; 32]), + 1.0, + true, + 6, + ), + ); + + manager + .handle_downstream_channel_request_in_aggregated_mode( + 7, + "new-miner".to_string(), + 1.0, + 6, + ) + .await + .unwrap(); + + let success = match sv1_server_receiver.try_recv().unwrap() { + MiningOwned::OpenExtendedMiningChannelSuccess(success) => success, + other => panic!("expected open success, got {other:?}"), + }; + assert_ne!(success.channel_id, AGGREGATED_CHANNEL_ID); + assert_eq!(success.channel_id, 1); + assert_eq!( + manager + .extended_channels + .with(&AGGREGATED_CHANNEL_ID, |channel| channel.get_channel_id()), + Some(42) + ); + } + + #[tokio::test] + async fn late_aggregated_join_targets_initial_job_to_new_channel_only() { + let (manager, sv1_server_receiver) = create_connected_aggregated_channel_manager(); + + manager + .extended_channels + .with_mut(&AGGREGATED_CHANNEL_ID, |aggregated_channel| { + aggregated_channel + .on_new_extended_mining_job(NewExtendedMiningJobOwned { + channel_id: 42, + job_id: 1, + min_ntime: Sv2OptionOwned::new(None), + version: 0x20000000, + version_rolling_allowed: true, + merkle_path: Seq0255Owned::new(vec![]).unwrap(), + coinbase_tx_prefix: hex::decode("02000000010000000000000000000000000000000000000000000000000000000000000000ffffffff265200162f5374726174756d2056322053524920506f6f6c2f2f08") + .unwrap() + .try_into() + .unwrap(), + coinbase_tx_suffix: hex::decode("feffffff0200f2052a01000000160014ebe1b7dcc293ccaa0ee743a86f89df8258c208fc0000000000000000266a24aa21a9ede2f61c3f71d1defd3fa999dfa36953755c690689799962b48bebd836974e8cf901000000") + .unwrap() + .try_into() + .unwrap(), + }) + .unwrap(); + aggregated_channel + .on_set_new_prev_hash(SetNewPrevHashOwned { + channel_id: 42, + job_id: 1, + prev_hash: vec![0; 32].try_into().unwrap(), + min_ntime: 0, + nbits: 0x207fffff, + }) + .unwrap(); + }) + .unwrap(); + + manager + .handle_downstream_channel_request_in_aggregated_mode( + 7, + "new-miner".to_string(), + 1.0, + 6, + ) + .await + .unwrap(); + + let success = match sv1_server_receiver.try_recv().unwrap() { + MiningOwned::OpenExtendedMiningChannelSuccess(success) => success, + other => panic!("expected open success, got {other:?}"), + }; + let job = match sv1_server_receiver.try_recv().unwrap() { + MiningOwned::NewExtendedMiningJob(job) => job, + other => panic!("expected initial job, got {other:?}"), + }; + + assert_eq!(job.channel_id, success.channel_id); + assert_ne!(job.channel_id, AGGREGATED_CHANNEL_ID); + } }