diff --git a/miner-apps/jd-client/src/lib/error.rs b/miner-apps/jd-client/src/lib/error.rs index 2e7eded04..9e4badbe7 100644 --- a/miner-apps/jd-client/src/lib/error.rs +++ b/miner-apps/jd-client/src/lib/error.rs @@ -41,6 +41,8 @@ use stratum_apps::{ }; use tokio::time::error::Elapsed; +use crate::config::ConfigJDCMode; + pub type JDCResult = Result>; #[derive(Debug)] @@ -58,6 +60,9 @@ pub struct Upstream; #[derive(Debug)] pub struct Downstream; +#[derive(Debug)] +pub struct JobDeclaratorClient; + #[derive(Debug)] pub struct JDCError { pub kind: JDCErrorKind, @@ -73,6 +78,12 @@ pub enum Action { Shutdown, } +impl Action { + pub fn is_shutdown(self) -> bool { + matches!(self, Self::Shutdown) + } +} + #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum LoopControl { Continue, @@ -85,12 +96,14 @@ impl CanDisconnect for ChannelManager {} impl CanFallback for Upstream {} impl CanFallback for JobDeclarator {} impl CanFallback for ChannelManager {} +impl CanFallback for JobDeclaratorClient {} impl CanShutdown for ChannelManager {} impl CanShutdown for TemplateProvider {} impl CanShutdown for Downstream {} impl CanShutdown for Upstream {} impl CanShutdown for JobDeclarator {} +impl CanShutdown for JobDeclaratorClient {} impl JDCError { pub fn log>(kind: E) -> Self { @@ -259,6 +272,12 @@ pub enum JDCErrorKind { InvalidKey, /// Upstream not found UpstreamNotFound, + /// Cannot determine Bitcoin data directory + InvalidBitcoinDataDir, + /// No upstream specified for pooled mining + NoUpstreamConfig(ConfigJDCMode), + /// Invalid coinbase output in config + InvalidCoinbaseOutput, } impl std::error::Error for JDCErrorKind {} @@ -402,6 +421,15 @@ impl fmt::Display for JDCErrorKind { CouldNotInitiateSystem => write!(f, "Could not initiate subsystem"), InvalidKey => write!(f, "Invalid key used during noise handshake"), UpstreamNotFound => write!(f, "Upstream not found"), + InvalidBitcoinDataDir => write!( + f, + "Could not determine Bitcoin data directory. Please set data_dir in config." + ), + NoUpstreamConfig(mode) => write!( + f, + "No upstreams configured for {mode:?} mode - at least one upstream is required" + ), + InvalidCoinbaseOutput => write!(f, "Invalid coinbase output in config"), } } } diff --git a/miner-apps/jd-client/src/lib/jdc_runtime.rs b/miner-apps/jd-client/src/lib/jdc_runtime.rs new file mode 100644 index 000000000..5cec5b0f0 --- /dev/null +++ b/miner-apps/jd-client/src/lib/jdc_runtime.rs @@ -0,0 +1,917 @@ +//! ## JDC Runtime Module +//! +//! Provides [`JdcRuntime`], a structured state-machine orchestrating the Job Declarator +//! Client's (JDC) initialization, bootstrap stages, background service loops, and graceful teardown +//! or fallback. + +use std::{sync::Arc, thread::JoinHandle, time::Duration}; + +use async_channel::{Receiver, Sender, unbounded}; +use stratum_apps::{ + bitcoin_core_sv2::CancellationToken, + fallback_coordinator::FallbackCoordinator, + stratum_core::{ + bitcoin::{TxOut, consensus::Encodable}, + parsers_sv2::{JobDeclaration, Mining, TemplateDistribution, Tlv}, + }, + task_manager::TaskManager, + tp_type::{TemplateProviderType, resolve_ipc_socket_path}, + utils::types::{DownstreamId, GRACEFUL_SHUTDOWN_TIMEOUT_SECONDS, Sv2Frame}, +}; +use tracing::{error, info, warn}; + +use crate::{ + JobDeclaratorClient, + channel_manager::ChannelManager, + config::ConfigJDCMode, + error::{Action, JDCError, JDCErrorKind, JDCResult}, + jd_mode::JDMode, + job_declarator::JobDeclarator, + template_receiver::{ + bitcoin_core::{BitcoinCoreSv2TDPConfig, connect_to_bitcoin_core}, + sv2_tp::Sv2Tp, + }, + upstream::Upstream, + utils::{UpstreamEntry, UpstreamState}, +}; +#[cfg(feature = "monitoring")] +use std::net::SocketAddr; + +struct Io { + channel_manager_to_upstream_sender: Sender, + channel_manager_to_upstream_receiver: Receiver, + upstream_to_channel_manager_sender: Sender, + upstream_to_channel_manager_receiver: Receiver, + channel_manager_to_jd_sender: Sender>, + channel_manager_to_jd_receiver: Receiver>, + jd_to_channel_manager_sender: Sender>, + jd_to_channel_manager_receiver: Receiver>, + downstream_to_channel_manager_sender: Sender<(DownstreamId, Mining<'static>, Option>)>, + downstream_to_channel_manager_receiver: + Receiver<(DownstreamId, Mining<'static>, Option>)>, + channel_manager_to_tp_sender: Sender>, + channel_manager_to_tp_receiver: Receiver>, + tp_to_channel_manager_sender: Sender>, + tp_to_channel_manager_receiver: Receiver>, +} + +struct BitcoinCoreSv2Handle { + join_handle: JoinHandle<()>, + cancellation_token: CancellationToken, +} + +/// The core coordinator of the JDC runtime, parameterized by its current bootstrap `State`. +/// +/// It manages the lifecycle of essential sub-services and channels, ensuring resources +/// are correctly initialized, passed to background executors, and cleanly torn down. +pub(super) struct JdcRuntime { + miner_coinbase_outputs: Vec, + encoded_outputs: Vec, + mode: JDMode, + fallback_coordinator: FallbackCoordinator, + task_manager: Arc, + bitcoin_core_sv2: Option, + jd_client: JobDeclaratorClient, + upstream_addresses: Vec, + state: State, +} + +pub(super) struct Init; + +struct IoReady { + io: Io, +} + +pub(super) struct TemplateProviderReady { + io: Io, +} + +struct ChannelManagerReady { + io: Io, + channel_manager: ChannelManager, +} + +struct UpstreamReady { + io: Io, + channel_manager: ChannelManager, +} + +struct SoloMiningReady { + io: Io, + channel_manager: ChannelManager, +} + +pub(super) struct Running { + io: Io, +} + +pub(super) struct Failed; + +pub(super) struct BootstrapError { + pub(super) kind: JDCErrorKind, + pub(super) runtime: JdcRuntime, +} + +impl From<(JDCErrorKind, JdcRuntime)> for BootstrapError { + fn from((kind, runtime): (JDCErrorKind, JdcRuntime)) -> Self { + Self { + kind, + runtime: runtime.into_failed(), + } + } +} + +impl JdcRuntime { + #[cfg(feature = "monitoring")] + fn start_monitoring_tasks( + &self, + channel_manager: &ChannelManager, + monitoring_addr: SocketAddr, + ) -> Result<(), String> { + let refresh_interval = Duration::from_secs( + self.jd_client + .config + .monitoring_cache_refresh_secs() + .unwrap_or(15), + ); + + let monitoring_server = stratum_apps::monitoring::MonitoringServer::new( + monitoring_addr, + Some(Arc::new(channel_manager.clone())), + Some(Arc::new(channel_manager.clone())), + refresh_interval, + ) + .map_err(|e| format!("failed to initialize monitoring server: {e}"))?; + + let cancellation_token_clone = self.jd_client.cancellation_token.clone(); + let fallback_coordinator_token = self.fallback_coordinator.token(); + let shutdown_signal = async move { + tokio::select! { + _ = cancellation_token_clone.cancelled() => { + info!("Monitoring server: received shutdown signal."); + } + _ = fallback_coordinator_token.cancelled() => { + info!("Monitoring server: fallback triggered."); + } + } + }; + + let monitoring_fallback = self.fallback_coordinator.clone(); + self.task_manager.spawn({ + let cancellation_token = self.jd_client.cancellation_token.clone(); + async move { + let fallback_handler = monitoring_fallback.register(); + + if let Err(e) = monitoring_server.run(shutdown_signal).await { + error!("Monitoring server error: {:?}", e); + cancellation_token.cancel(); + } + + fallback_handler.done(); + info!("Monitoring server task exited and signaled fallback coordinator"); + } + }); + + let telemetry_fallback = self.fallback_coordinator.clone(); + let telemetry_cm = channel_manager.clone(); + self.task_manager.spawn({ + let cancellation_token = self.jd_client.cancellation_token.clone(); + async move { + let fallback_token = telemetry_fallback.token(); + let fallback_handler = telemetry_fallback.register(); + + telemetry_cm + .run_miner_telemetry_loop(refresh_interval, cancellation_token, fallback_token) + .await; + + fallback_handler.done(); + info!("JDC miner telemetry task exited and signaled fallback coordinator"); + } + }); + + Ok(()) + } + + async fn start_services_inner(&self, io: &Io, channel_manager: ChannelManager) { + // Start monitoring server if configured + #[cfg(feature = "monitoring")] + if let Some(monitoring_addr) = self.jd_client.config.monitoring_address() { + info!("Initializing monitoring server on http://{monitoring_addr}"); + if let Err(e) = self.start_monitoring_tasks(&channel_manager, monitoring_addr) { + error!("Failed to initialize monitoring tasks: {e}"); + self.jd_client.cancellation_token.cancel(); + } + } + + self.task_manager.spawn({ + let config = self.jd_client.config.clone(); + let cancellation_token = self.jd_client.cancellation_token.clone(); + let task_manager = self.task_manager.clone(); + let fallback_coordinator = self.fallback_coordinator.clone(); + let downstream_to_channel_manager_sender = + io.downstream_to_channel_manager_sender.clone(); + async move { + if let Err(e) = channel_manager + .start_downstream_server( + *config.authority_public_key(), + *config.authority_secret_key(), + config.cert_validity_sec(), + *config.listening_address(), + task_manager, + cancellation_token.clone(), + fallback_coordinator, + downstream_to_channel_manager_sender, + config.supported_extensions().to_vec(), + config.required_extensions().to_vec(), + ) + .await + { + tracing::error!(?e, "Downstream server task exited with error"); + cancellation_token.cancel(); + } + } + }); + } + + fn into_failed(self) -> JdcRuntime { + JdcRuntime { + miner_coinbase_outputs: self.miner_coinbase_outputs, + encoded_outputs: self.encoded_outputs, + mode: self.mode, + fallback_coordinator: self.fallback_coordinator, + task_manager: self.task_manager, + bitcoin_core_sv2: self.bitcoin_core_sv2, + jd_client: self.jd_client, + upstream_addresses: self.upstream_addresses, + state: Failed, + } + } + + /// Performs a coordinated, graceful shutdown of the runtime. + /// + /// Signals cancellation to all active sub-services and background tasks, awaiting + /// their clean termination up to a configured graceful timeout. + pub async fn shutdown(self) { + self.jd_client.cancellation_token.cancel(); + + if let Some(bitcoin_core_sv2) = self.bitcoin_core_sv2 { + bitcoin_core_sv2.cancellation_token.cancel(); + + info!("Waiting for BitcoinCoreSv2TDP dedicated thread to shutdown..."); + match bitcoin_core_sv2.join_handle.join() { + Ok(_) => info!("BitcoinCoreSv2TDP dedicated thread shutdown complete."), + Err(e) => error!("BitcoinCoreSv2TDP dedicated thread error: {e:?}"), + } + } + + warn!( + "Graceful shutdown: waiting {} seconds for tasks to finish", + GRACEFUL_SHUTDOWN_TIMEOUT_SECONDS + ); + match tokio::time::timeout( + std::time::Duration::from_secs(GRACEFUL_SHUTDOWN_TIMEOUT_SECONDS), + self.task_manager.join_all(), + ) + .await + { + Ok(_) => { + info!("All tasks joined cleanly"); + } + Err(_) => { + warn!( + "Tasks did not finish within {} seconds, aborting", + GRACEFUL_SHUTDOWN_TIMEOUT_SECONDS + ); + self.task_manager.abort_all().await; + info!("Joining aborted tasks..."); + self.task_manager.join_all().await; + warn!("Forced shutdown complete"); + } + } + self.jd_client.mark_stopped(); + info!("JD Client shutdown complete."); + } +} + +impl JdcRuntime { + pub(super) fn new(jd_client: JobDeclaratorClient) -> Result { + let miner_coinbase_outputs = vec![jd_client.config.get_txout()]; + let mut encoded_outputs = vec![]; + let mode = JDMode::new(jd_client.config.mode); + let upstream_addresses = jd_client + .config + .upstreams() + .iter() + .map(|u| UpstreamEntry { + pool_host: u.pool_address.clone(), + pool_port: u.pool_port, + jds_host: u.jds_address.clone(), + jds_port: u.jds_port, + authority_pubkey: u.authority_pubkey, + tried_or_flagged: false, + user_identity: u.user_identity.clone(), + }) + .collect(); + + if miner_coinbase_outputs + .consensus_encode(&mut encoded_outputs) + .is_err() + { + return Err(JDCErrorKind::InvalidCoinbaseOutput); + } + + Ok(JdcRuntime { + miner_coinbase_outputs, + encoded_outputs, + mode, + fallback_coordinator: FallbackCoordinator::new(), + task_manager: Arc::new(TaskManager::new()), + bitcoin_core_sv2: None, + jd_client, + upstream_addresses, + state: Init, + }) + } + + /// Drives the linear bootstrap sequence of the JDC, transitioning the runtime + /// from [`Init`] to the active [`Running`] state. + /// + /// If an intermediate phase fails, the caller receives the partially initialized + /// runtime and is responsible for shutting down any resources that were already started. + pub async fn bootstrap(self) -> Result, BootstrapError> { + let runtime = self.bootstrap_io(); + + let runtime = runtime.bootstrap_template_provider().await?; + + runtime.bootstrap_mining().await + } + + fn bootstrap_io(self) -> JdcRuntime { + let (channel_manager_to_upstream_sender, channel_manager_to_upstream_receiver) = + unbounded(); + let (upstream_to_channel_manager_sender, upstream_to_channel_manager_receiver) = + unbounded(); + + let (channel_manager_to_jd_sender, channel_manager_to_jd_receiver) = unbounded(); + let (jd_to_channel_manager_sender, jd_to_channel_manager_receiver) = unbounded(); + + let (downstream_to_channel_manager_sender, downstream_to_channel_manager_receiver) = + unbounded(); + + let (channel_manager_to_tp_sender, channel_manager_to_tp_receiver) = unbounded(); + let (tp_to_channel_manager_sender, tp_to_channel_manager_receiver) = unbounded(); + + JdcRuntime { + miner_coinbase_outputs: self.miner_coinbase_outputs, + encoded_outputs: self.encoded_outputs, + mode: self.mode, + fallback_coordinator: self.fallback_coordinator, + task_manager: self.task_manager, + bitcoin_core_sv2: self.bitcoin_core_sv2, + jd_client: self.jd_client, + upstream_addresses: self.upstream_addresses, + state: IoReady { + io: Io { + channel_manager_to_upstream_sender, + channel_manager_to_upstream_receiver, + upstream_to_channel_manager_sender, + upstream_to_channel_manager_receiver, + channel_manager_to_jd_sender, + channel_manager_to_jd_receiver, + jd_to_channel_manager_sender, + jd_to_channel_manager_receiver, + downstream_to_channel_manager_sender, + downstream_to_channel_manager_receiver, + channel_manager_to_tp_sender, + channel_manager_to_tp_receiver, + tp_to_channel_manager_sender, + tp_to_channel_manager_receiver, + }, + }, + } + } +} + +impl JdcRuntime { + async fn bootstrap_template_provider( + self, + ) -> Result, (JDCErrorKind, JdcRuntime)> { + let mut bitcoin_core_sv2: Option = None; + + match self.jd_client.config.template_provider_type().clone() { + TemplateProviderType::Sv2Tp { + address, + public_key, + } => { + let template_receiver = match Sv2Tp::new( + address.clone(), + public_key, + self.state.io.channel_manager_to_tp_receiver.clone(), + self.state.io.tp_to_channel_manager_sender.clone(), + self.jd_client.cancellation_token.clone(), + self.task_manager.clone(), + ) + .await + { + Ok(template_receiver) => template_receiver, + Err(e) => { + error!(error = ?e, "Failed to initialize SV2 template receiver"); + return Err((e.kind, self)); + } + }; + + let cancellation_token_tp = self.jd_client.cancellation_token.clone(); + let task_manager_cl = self.task_manager.clone(); + + if let Err(e) = template_receiver + .start(address, cancellation_token_tp, task_manager_cl) + .await + { + error!(error = ?e, "Failed to start SV2 template receiver"); + return Err((e.kind, self)); + } + + info!("Sv2 Template Provider setup done"); + } + TemplateProviderType::BitcoinCoreIpc { + version, + network, + data_dir, + fee_threshold, + min_interval, + } => { + let unix_socket_path = match resolve_ipc_socket_path(&network, data_dir) { + Some(unix_socket_path) => unix_socket_path, + None => { + error!( + "Could not determine Bitcoin data directory. Please set data_dir in config." + ); + return Err((JDCErrorKind::InvalidBitcoinDataDir, self)); + } + }; + + info!( + "Using Bitcoin Core IPC socket at: {}", + unix_socket_path.display() + ); + + // incoming and outgoing TDP channels from the perspective of BitcoinCoreSv2TDP + let incoming_tdp_receiver = self.state.io.channel_manager_to_tp_receiver.clone(); + let outgoing_tdp_sender = self.state.io.tp_to_channel_manager_sender.clone(); + + let bitcoin_core_cancellation_token = CancellationToken::new(); + let bitcoin_core_config = BitcoinCoreSv2TDPConfig { + version, + unix_socket_path, + fee_threshold, + min_interval, + incoming_tdp_receiver, + outgoing_tdp_sender, + cancellation_token: bitcoin_core_cancellation_token.clone(), + }; + + bitcoin_core_sv2 = Some(BitcoinCoreSv2Handle { + join_handle: connect_to_bitcoin_core( + bitcoin_core_config, + self.jd_client.cancellation_token.clone(), + self.task_manager.clone(), + ) + .await, + cancellation_token: bitcoin_core_cancellation_token, + }); + } + } + + Ok(JdcRuntime { + miner_coinbase_outputs: self.miner_coinbase_outputs, + encoded_outputs: self.encoded_outputs, + mode: self.mode, + fallback_coordinator: self.fallback_coordinator, + task_manager: self.task_manager, + bitcoin_core_sv2, + jd_client: self.jd_client, + upstream_addresses: self.upstream_addresses, + state: TemplateProviderReady { io: self.state.io }, + }) + } +} + +impl JdcRuntime { + pub async fn bootstrap_mining(self) -> Result, BootstrapError> { + let runtime = self.bootstrap_channel_manager().await?; + + if runtime.jd_client.config.mode == ConfigJDCMode::SoloMining { + return Ok(runtime.into_solo_mining().start_services().await); + } + + match runtime.try_upstream().await { + Ok(upstream_ready) => Ok(upstream_ready.start_services().await), + Err((JDCErrorKind::NoUpstreamConfig(mode), runtime)) => { + error!( + ?mode, + "Invalid configuration: JDC mode is non-solo ({:?}) but no upstream server addresses were configured", + mode + ); + Err((JDCErrorKind::NoUpstreamConfig(mode), runtime).into()) + } + Err((_, template_provider_ready)) => { + info!("Upstream initialization failed; falling back to solo mining mode"); + Ok(template_provider_ready + .into_solo_mining() + .start_services() + .await) + } + } + } + + async fn bootstrap_channel_manager( + self, + ) -> Result, (JDCErrorKind, JdcRuntime)> + { + let channel_manager = match ChannelManager::new( + self.jd_client.config.clone(), + self.state.io.channel_manager_to_upstream_sender.clone(), + self.state.io.upstream_to_channel_manager_receiver.clone(), + self.state.io.channel_manager_to_jd_sender.clone(), + self.state.io.jd_to_channel_manager_receiver.clone(), + self.state.io.channel_manager_to_tp_sender.clone(), + self.state.io.tp_to_channel_manager_receiver.clone(), + self.state.io.downstream_to_channel_manager_receiver.clone(), + self.encoded_outputs.clone(), + self.jd_client.config.supported_extensions().to_vec(), + self.jd_client.config.required_extensions().to_vec(), + self.mode.clone(), + ) + .await + { + Ok(cm) => cm, + Err(e) => return Err((e.kind, self)), + }; + + channel_manager + .clone() + .start( + self.jd_client.cancellation_token.clone(), + self.fallback_coordinator.clone(), + self.task_manager.clone(), + self.miner_coinbase_outputs.clone(), + ) + .await; + + Ok(JdcRuntime { + miner_coinbase_outputs: self.miner_coinbase_outputs, + encoded_outputs: self.encoded_outputs, + mode: self.mode, + fallback_coordinator: self.fallback_coordinator, + task_manager: self.task_manager, + bitcoin_core_sv2: self.bitcoin_core_sv2, + jd_client: self.jd_client, + upstream_addresses: self.upstream_addresses, + state: ChannelManagerReady { + io: self.state.io, + channel_manager, + }, + }) + } +} + +impl JdcRuntime { + async fn try_upstream( + mut self, + ) -> Result, (JDCErrorKind, JdcRuntime)> { + if self.upstream_addresses.is_empty() { + return Err(( + JDCErrorKind::NoUpstreamConfig(self.jd_client.config.mode), + self, + )); + } + + info!("Attempting to initialize upstream..."); + + match self.initialize_jd().await { + Ok((upstream, job_declarator, user_identity)) => { + upstream + .start( + self.jd_client.config.min_supported_version(), + self.jd_client.config.max_supported_version(), + self.jd_client.cancellation_token.clone(), + self.fallback_coordinator.clone(), + self.task_manager.clone(), + ) + .await; + + job_declarator + .start( + self.jd_client.cancellation_token.clone(), + self.fallback_coordinator.clone(), + self.task_manager.clone(), + ) + .await; + + self.state.channel_manager.set_user_identity(user_identity); + self.state + .channel_manager + .upstream_state + .set(UpstreamState::NoChannel); + _ = self.state.channel_manager.allocate_tokens(2).await; + + Ok(JdcRuntime { + miner_coinbase_outputs: self.miner_coinbase_outputs, + encoded_outputs: self.encoded_outputs, + mode: self.mode, + fallback_coordinator: self.fallback_coordinator, + task_manager: self.task_manager, + bitcoin_core_sv2: self.bitcoin_core_sv2, + jd_client: self.jd_client, + upstream_addresses: self.upstream_addresses, + state: UpstreamReady { + io: self.state.io, + channel_manager: self.state.channel_manager, + }, + }) + } + Err(e) => { + tracing::error!("Failed to initialize upstream: {:?}", e); + Err((e, self)) + } + } + } + + fn into_solo_mining(self) -> JdcRuntime { + info!("Starting in solo mining mode"); + self.mode.set_solo_mining(); + + JdcRuntime { + miner_coinbase_outputs: self.miner_coinbase_outputs, + encoded_outputs: self.encoded_outputs, + mode: self.mode, + fallback_coordinator: self.fallback_coordinator, + task_manager: self.task_manager, + bitcoin_core_sv2: self.bitcoin_core_sv2, + jd_client: self.jd_client, + upstream_addresses: self.upstream_addresses, + state: SoloMiningReady { + io: self.state.io, + channel_manager: self.state.channel_manager, + }, + } + } + + /// Initializes an upstream pool + JD connection pair. + async fn initialize_jd(&mut self) -> Result<(Upstream, JobDeclarator, String), JDCErrorKind> { + const MAX_RETRIES: usize = 3; + let upstream_len = self.upstream_addresses.len(); + + for i in 0..upstream_len { + let upstream_entry = self.upstream_addresses[i].clone(); + info!( + "Trying upstream {} of {}: pool={}:{}, jds={}:{}", + i + 1, + upstream_len, + upstream_entry.pool_host, + upstream_entry.pool_port, + upstream_entry.jds_host, + upstream_entry.jds_port, + ); + + tokio::select! { + biased; + _ = self.jd_client.cancellation_token.cancelled() => { + info!("Shutdown requested while waiting to initialize upstream, aborting retries"); + return Err(JDCErrorKind::CouldNotInitiateSystem); + } + _ = tokio::time::sleep(Duration::from_secs(1)) => {} + } + + if upstream_entry.tried_or_flagged { + info!( + "Upstream previously marked as malicious, skipping initial attempt warnings." + ); + continue; + } + + for attempt in 1..=MAX_RETRIES { + if self.jd_client.cancellation_token.is_cancelled() { + info!( + "Shutdown requested before upstream connection attempt, aborting retries" + ); + return Err(JDCErrorKind::CouldNotInitiateSystem); + } + + info!("Connection attempt {}/{}...", attempt, MAX_RETRIES); + + match self.try_initialize_single(&upstream_entry).await { + Ok((upstream, jd)) => { + self.upstream_addresses[i].tried_or_flagged = true; + return Ok((upstream, jd, upstream_entry.user_identity.clone())); + } + Err(e) => { + tracing::error!("Upstream and JDS connection terminated"); + + tokio::select! { + biased; + _ = self.jd_client.cancellation_token.cancelled() => { + info!("Shutdown requested after upstream initialization failure, aborting retries"); + return Err(JDCErrorKind::CouldNotInitiateSystem); + } + _ = tokio::time::sleep(Duration::from_secs(1)) => {} + } + + // Stop retrying and fail immediately if a shutdown signal is encountered, + // as retries are only intended for fallback scenarios. + if e.action.is_shutdown() { + info!( + "Encountered a shutdown error during upstream initialization, aborting retries" + ); + return Err(e.kind); + } + + warn!( + "Attempt {}/{} failed for pool={}:{}, jds={}:{}: {:?}", + attempt, + MAX_RETRIES, + upstream_entry.pool_host, + upstream_entry.pool_port, + upstream_entry.jds_host, + upstream_entry.jds_port, + e + ); + if attempt == MAX_RETRIES { + warn!( + "Max retries reached for pool={}:{}, jds={}:{}, moving to next upstream", + upstream_entry.pool_host, + upstream_entry.pool_port, + upstream_entry.jds_host, + upstream_entry.jds_port, + ); + } + } + } + } + self.upstream_addresses[i].tried_or_flagged = true; + } + + tracing::error!("All upstreams failed after {} retries each", MAX_RETRIES); + Err(JDCErrorKind::CouldNotInitiateSystem) + } + + // Attempts to initialize a single upstream (pool + JDS pair). + #[cfg_attr(not(test), hotpath::measure)] + async fn try_initialize_single( + &self, + upstream_entry: &UpstreamEntry, + ) -> JDCResult<(Upstream, JobDeclarator), super::error::JobDeclaratorClient> { + info!("Upstream connection in-progress at initialize single"); + let upstream = Upstream::new( + upstream_entry, + self.state.io.upstream_to_channel_manager_sender.clone(), + self.state.io.channel_manager_to_upstream_receiver.clone(), + self.jd_client.cancellation_token.clone(), + self.fallback_coordinator.clone(), + self.task_manager.clone(), + self.jd_client.config.required_extensions().to_vec(), + ) + .await + .map_err(|err| match err.action { + Action::Shutdown => JDCError::shutdown(err.kind), + _ => JDCError::fallback(err.kind), + })?; + + info!("Upstream connection done at initialize single"); + + let job_declarator = JobDeclarator::new( + upstream_entry, + self.state.io.jd_to_channel_manager_sender.clone(), + self.state.io.channel_manager_to_jd_receiver.clone(), + self.jd_client.cancellation_token.clone(), + self.fallback_coordinator.clone(), + self.mode.clone(), + self.task_manager.clone(), + ) + .await + .map_err(|err| match err.action { + Action::Shutdown => JDCError::shutdown(err.kind), + _ => JDCError::fallback(err.kind), + })?; + + Ok((upstream, job_declarator)) + } +} + +impl JdcRuntime { + /// Activates the background execution loops of the [`ChannelManager`], downstream server, and + /// monitoring server, transitioning the runtime to [`Running`]. + async fn start_services(self) -> JdcRuntime { + self.start_services_inner(&self.state.io, self.state.channel_manager.clone()) + .await; + + JdcRuntime { + miner_coinbase_outputs: self.miner_coinbase_outputs, + encoded_outputs: self.encoded_outputs, + mode: self.mode, + fallback_coordinator: self.fallback_coordinator, + task_manager: self.task_manager, + bitcoin_core_sv2: self.bitcoin_core_sv2, + jd_client: self.jd_client, + upstream_addresses: self.upstream_addresses, + state: Running { io: self.state.io }, + } + } +} + +impl JdcRuntime { + /// Activates the background execution loops of the [`ChannelManager`], downstream server, and + /// monitoring server, transitioning the runtime to [`Running`]. + async fn start_services(self) -> JdcRuntime { + self.start_services_inner(&self.state.io, self.state.channel_manager.clone()) + .await; + + JdcRuntime { + miner_coinbase_outputs: self.miner_coinbase_outputs, + encoded_outputs: self.encoded_outputs, + mode: self.mode, + fallback_coordinator: self.fallback_coordinator, + task_manager: self.task_manager, + bitcoin_core_sv2: self.bitcoin_core_sv2, + jd_client: self.jd_client, + upstream_addresses: self.upstream_addresses, + state: Running { io: self.state.io }, + } + } +} + +pub enum RuntimeEvent { + Shutdown, + Fallback, +} + +impl JdcRuntime { + pub async fn wait(&self) -> RuntimeEvent { + info!("Spawning status listener task..."); + let fallback_token = self.fallback_coordinator.token(); + + tokio::select! { + biased; + + _ = self.jd_client.cancellation_token.cancelled() => { + RuntimeEvent::Shutdown + } + _ = fallback_token.cancelled() => { + warn!("Upstream/Job Declarator connection dropped — attempting reconnection..."); + RuntimeEvent::Fallback + } + _ = tokio::signal::ctrl_c() => { + info!("Ctrl+C received — initiating graceful shutdown..."); + RuntimeEvent::Shutdown + } + } + } + + pub async fn cleanup_for_fallback(self) -> JdcRuntime { + // trigger fallback and wait for all components to finish cleanup + self.fallback_coordinator.trigger_fallback_and_wait().await; + info!("All components finished fallback cleanup"); + + self.mode.set_solo_mining(); + info!("Existing Upstream or JD instance taken out. Preparing fallback."); + + let (channel_manager_to_upstream_sender, channel_manager_to_upstream_receiver) = + unbounded(); + let (upstream_to_channel_manager_sender, upstream_to_channel_manager_receiver) = + unbounded(); + let (channel_manager_to_jd_sender, channel_manager_to_jd_receiver) = unbounded(); + let (jd_to_channel_manager_sender, jd_to_channel_manager_receiver) = unbounded(); + let (downstream_to_channel_manager_sender, downstream_to_channel_manager_receiver) = + unbounded(); + + let new_io = Io { + channel_manager_to_upstream_sender, + channel_manager_to_upstream_receiver, + upstream_to_channel_manager_sender, + upstream_to_channel_manager_receiver, + channel_manager_to_jd_sender, + channel_manager_to_jd_receiver, + jd_to_channel_manager_sender, + jd_to_channel_manager_receiver, + downstream_to_channel_manager_sender, + downstream_to_channel_manager_receiver, + // Re-use TP channels from running session + channel_manager_to_tp_sender: self.state.io.channel_manager_to_tp_sender, + channel_manager_to_tp_receiver: self.state.io.channel_manager_to_tp_receiver, + tp_to_channel_manager_sender: self.state.io.tp_to_channel_manager_sender, + tp_to_channel_manager_receiver: self.state.io.tp_to_channel_manager_receiver, + }; + + JdcRuntime { + miner_coinbase_outputs: self.miner_coinbase_outputs, + encoded_outputs: self.encoded_outputs, + mode: self.mode, + fallback_coordinator: FallbackCoordinator::new(), + task_manager: self.task_manager, + bitcoin_core_sv2: self.bitcoin_core_sv2, + jd_client: self.jd_client, + upstream_addresses: self.upstream_addresses, + state: TemplateProviderReady { io: new_io }, + } + } +} diff --git a/miner-apps/jd-client/src/lib/mod.rs b/miner-apps/jd-client/src/lib/mod.rs index 8d7b58cea..8322a627e 100644 --- a/miner-apps/jd-client/src/lib/mod.rs +++ b/miner-apps/jd-client/src/lib/mod.rs @@ -1,39 +1,15 @@ -#[cfg(feature = "monitoring")] -use std::net::SocketAddr; -use std::{ - sync::{ - Arc, - atomic::{AtomicBool, Ordering}, - }, - thread::JoinHandle, - time::Duration, +use std::sync::{ + Arc, + atomic::{AtomicBool, Ordering}, }; -use async_channel::{Receiver, Sender, unbounded}; -use stratum_apps::{ - bitcoin_core_sv2::CancellationToken, - fallback_coordinator::FallbackCoordinator, - stratum_core::{bitcoin::consensus::Encodable, parsers_sv2::JobDeclaration}, - task_manager::TaskManager, - tp_type::TemplateProviderType, - utils::types::{GRACEFUL_SHUTDOWN_TIMEOUT_SECONDS, Sv2Frame}, -}; +use error::JDCErrorKind; +use jdc_runtime::{Init, JdcRuntime, RuntimeEvent}; +use stratum_apps::bitcoin_core_sv2::CancellationToken; use tokio::sync::Notify; -use tracing::{debug, error, info, warn}; +use tracing::{error, info}; -use crate::{ - channel_manager::ChannelManager, - config::JobDeclaratorClientConfig, - error::JDCErrorKind, - jd_mode::JDMode, - job_declarator::JobDeclarator, - template_receiver::{ - bitcoin_core::{BitcoinCoreSv2TDPConfig, connect_to_bitcoin_core}, - sv2_tp::Sv2Tp, - }, - upstream::Upstream, - utils::{UpstreamEntry, UpstreamState}, -}; +use crate::config::JobDeclaratorClientConfig; mod channel_manager; pub mod config; @@ -41,6 +17,7 @@ mod downstream; pub mod error; mod io_task; pub mod jd_mode; +mod jdc_runtime; mod job_declarator; #[cfg(feature = "monitoring")] pub mod monitoring; @@ -69,577 +46,63 @@ impl JobDeclaratorClient { } } - /// Starts the Job Declarator Client (JDC) main loop. - pub async fn start(&self) { - info!("Job declarator client starting... setting up subsystems"); - - let miner_coinbase_outputs = vec![self.config.get_txout()]; - let mut encoded_outputs = vec![]; - let mode = JDMode::new(self.config.mode); - - if let Err(e) = miner_coinbase_outputs.consensus_encode(&mut encoded_outputs) { - error!(error = ?e, "Invalid coinbase output in config"); - self.cancellation_token.cancel(); - self.shutdown_notify.notify_waiters(); - self.is_alive.store(false, Ordering::Relaxed); - return; - } - - let mut fallback_coordinator = FallbackCoordinator::new(); - let task_manager = Arc::new(TaskManager::new()); - - let (channel_manager_to_upstream_sender, channel_manager_to_upstream_receiver) = - unbounded(); - let (upstream_to_channel_manager_sender, upstream_to_channel_manager_receiver) = - unbounded(); - - let (channel_manager_to_jd_sender, channel_manager_to_jd_receiver) = unbounded(); - let (jd_to_channel_manager_sender, jd_to_channel_manager_receiver) = unbounded(); - - let (downstream_to_channel_manager_sender, downstream_to_channel_manager_receiver) = - unbounded(); - - let (channel_manager_to_tp_sender, channel_manager_to_tp_receiver) = unbounded(); - let (tp_to_channel_manager_sender, tp_to_channel_manager_receiver) = unbounded(); - - debug!("Channels initialized."); + pub fn mark_stopped(&self) { + self.is_alive.store(false, Ordering::Release); + self.cancellation_token.cancel(); + self.shutdown_notify.notify_waiters(); + } - let channel_manager = match ChannelManager::new( - self.config.clone(), - channel_manager_to_upstream_sender.clone(), - upstream_to_channel_manager_receiver.clone(), - channel_manager_to_jd_sender.clone(), - jd_to_channel_manager_receiver.clone(), - channel_manager_to_tp_sender.clone(), - tp_to_channel_manager_receiver.clone(), - downstream_to_channel_manager_receiver, - encoded_outputs.clone(), - self.config.supported_extensions().to_vec(), - self.config.required_extensions().to_vec(), - mode.clone(), - ) - .await - { - Ok(channel_manager) => channel_manager, + /// Starts the main event loop for the Job Declarator Client (JDC). + /// + /// The startup and execution sequence follows: + /// 1. **Initialize:** Sets up the JDC runtime state machine ([`JdcRuntime`]). + /// 2. **Bootstrap:** Configures internal channels, connects to the Template Provider, + /// initializes the Channel Manager, and establishes upstream SV2 connections or solo mining. + /// 3. **Run & Loop:** Spawns active background loops/servers and handles fallbacks or graceful + /// shutdown. + /// 4. **Teardown:** Performs a coordinated graceful cleanup of all services and tasks upon + /// termination. + pub async fn start(&self) -> Result<(), JDCErrorKind> { + let runtime = match JdcRuntime::::new(self.clone()) { + Ok(runtime) => runtime, Err(e) => { - error!(error = ?e, "Failed to initialize channel manager"); - self.cancellation_token.cancel(); - self.shutdown_notify.notify_waiters(); - self.is_alive.store(false, Ordering::Relaxed); - return; + self.mark_stopped(); + return Err(e); } }; - // Start monitoring server if configured - #[cfg(feature = "monitoring")] - if let Some(monitoring_addr) = self.config.monitoring_address() { - info!("Initializing monitoring server on http://{monitoring_addr}"); - if let Err(e) = self.start_monitoring_tasks( - monitoring_addr, - channel_manager.clone(), - self.cancellation_token.clone(), - fallback_coordinator.clone(), - task_manager.clone(), - ) { - error!("Failed to initialize monitoring tasks: {e}"); - self.cancellation_token.cancel(); + let mut running = match runtime.bootstrap().await { + Ok(running) => running, + Err(bootstrap_err) => { + error!(?bootstrap_err.kind, "Failed to bootstrap JDC"); + bootstrap_err.runtime.shutdown().await; + self.mark_stopped(); + return Err(bootstrap_err.kind); } - } - - let initial_channel_manager = channel_manager.clone(); - let mut bitcoin_core_sv2_join_handle: Option> = None; - let mut bitcoin_core_sv2_cancellation_token: Option = None; - - match self.config.template_provider_type().clone() { - TemplateProviderType::Sv2Tp { - address, - public_key, - } => { - let template_receiver = match Sv2Tp::new( - address.clone(), - public_key, - channel_manager_to_tp_receiver, - tp_to_channel_manager_sender, - self.cancellation_token.clone(), - task_manager.clone(), - ) - .await - { - Ok(template_receiver) => template_receiver, - Err(e) => { - error!(error = ?e, "Failed to initialize SV2 template receiver"); - self.cancellation_token.cancel(); - self.shutdown_notify.notify_waiters(); - self.is_alive.store(false, Ordering::Relaxed); - return; - } - }; - - let cancellation_token_tp = self.cancellation_token.clone(); - let task_manager_cl = task_manager.clone(); - - if let Err(e) = template_receiver - .start(address, cancellation_token_tp, task_manager_cl) - .await - { - error!(error = ?e, "Failed to start SV2 template receiver"); - self.cancellation_token.cancel(); - self.shutdown_notify.notify_waiters(); - self.is_alive.store(false, Ordering::Relaxed); - return; - } - - info!("Sv2 Template Provider setup done"); - } - TemplateProviderType::BitcoinCoreIpc { - version, - network, - data_dir, - fee_threshold, - min_interval, - } => { - let unix_socket_path = match stratum_apps::tp_type::resolve_ipc_socket_path( - &network, data_dir, - ) { - Some(unix_socket_path) => unix_socket_path, - None => { - error!( - "Could not determine Bitcoin data directory. Please set data_dir in config." - ); - self.cancellation_token.cancel(); - self.shutdown_notify.notify_waiters(); - self.is_alive.store(false, Ordering::Relaxed); - return; - } - }; - - info!( - "Using Bitcoin Core IPC socket at: {}", - unix_socket_path.display() - ); - - // incoming and outgoing TDP channels from the perspective of BitcoinCoreSv2TDP - let incoming_tdp_receiver = channel_manager_to_tp_receiver.clone(); - let outgoing_tdp_sender = tp_to_channel_manager_sender.clone(); - - let bitcoin_core_cancellation_token = CancellationToken::new(); - let bitcoin_core_config = BitcoinCoreSv2TDPConfig { - version, - unix_socket_path, - fee_threshold, - min_interval, - incoming_tdp_receiver, - outgoing_tdp_sender, - cancellation_token: bitcoin_core_cancellation_token.clone(), - }; - - bitcoin_core_sv2_cancellation_token = Some(bitcoin_core_cancellation_token); - bitcoin_core_sv2_join_handle = Some( - connect_to_bitcoin_core( - bitcoin_core_config, - self.cancellation_token.clone(), - task_manager.clone(), - ) - .await, - ); - } - } - - let mut upstream_addresses: Vec<_> = self - .config - .upstreams() - .iter() - .map(|u| UpstreamEntry { - pool_host: u.pool_address.clone(), - pool_port: u.pool_port, - jds_host: u.jds_address.clone(), - jds_port: u.jds_port, - authority_pubkey: u.authority_pubkey, - tried_or_flagged: false, - user_identity: u.user_identity.clone(), - }) - .collect(); - - channel_manager - .clone() - .start( - self.cancellation_token.clone(), - fallback_coordinator.clone(), - task_manager.clone(), - miner_coinbase_outputs.clone(), - ) - .await; - - if self.config.mode == config::ConfigJDCMode::SoloMining { - if !upstream_addresses.is_empty() { - warn!( - "Solo mining mode configured but upstreams are present - they will be ignored" - ); - } - info!("Starting in solo mining mode"); - mode.set_solo_mining(); - } else if upstream_addresses.is_empty() { - error!( - "No upstreams configured for {:?} mode - at least one upstream is required", - self.config.mode - ); - self.cancellation_token.cancel(); - } else { - info!("Attempting to initialize upstream..."); - - match self - .initialize_jd( - &mut upstream_addresses, - channel_manager_to_upstream_receiver.clone(), - upstream_to_channel_manager_sender.clone(), - channel_manager_to_jd_receiver.clone(), - jd_to_channel_manager_sender.clone(), - self.cancellation_token.clone(), - fallback_coordinator.clone(), - mode.clone(), - task_manager.clone(), - ) - .await - { - Ok((upstream, job_declarator, user_identity)) => { - initial_channel_manager.set_user_identity(user_identity); - - upstream - .start( - self.config.min_supported_version(), - self.config.max_supported_version(), - self.cancellation_token.clone(), - fallback_coordinator.clone(), - task_manager.clone(), - ) - .await; - - job_declarator - .start( - self.cancellation_token.clone(), - fallback_coordinator.clone(), - task_manager.clone(), - ) - .await; - - initial_channel_manager - .upstream_state - .set(UpstreamState::NoChannel); - _ = initial_channel_manager.allocate_tokens(2).await; - } - Err(e) => { - tracing::error!("Failed to initialize upstream: {:?}", e); - mode.set_solo_mining(); - } - }; - } - - task_manager.spawn({ - let config = self.config.clone(); - let cancellation_token = self.cancellation_token.clone(); - let task_manager = task_manager.clone(); - let fallback_coordinator = fallback_coordinator.clone(); - async move { - if let Err(e) = initial_channel_manager - .start_downstream_server( - *config.authority_public_key(), - *config.authority_secret_key(), - config.cert_validity_sec(), - *config.listening_address(), - task_manager, - cancellation_token.clone(), - fallback_coordinator, - downstream_to_channel_manager_sender, - config.supported_extensions().to_vec(), - config.required_extensions().to_vec(), - ) - .await - { - tracing::error!(?e, "Downstream server task exited with error"); - cancellation_token.cancel(); - } - } - }); - - info!("Spawning status listener task..."); - let mut fallback_token = fallback_coordinator.token(); + }; loop { - tokio::select! { - biased; - - _ = self.cancellation_token.cancelled() => { - break; + match running.wait().await { + RuntimeEvent::Shutdown => { + running.shutdown().await; + self.mark_stopped(); + return Ok(()); } - _ = fallback_token.cancelled() => { - warn!("Upstream/Job Declarator connection dropped — attempting reconnection..."); - - // trigger fallback and wait for all components to finish cleanup - fallback_coordinator.trigger_fallback_and_wait().await; - info!("All components finished fallback cleanup"); - - mode.set_solo_mining(); - info!("Existing Upstream or JD instance taken out. Preparing fallback."); - - // Create a fresh FallbackCoordinator for the reconnection attempt - fallback_coordinator = FallbackCoordinator::new(); - fallback_token = fallback_coordinator.token(); - - // Recreate channels (old ones were closed during fallback) - let (channel_manager_to_upstream_sender_new, channel_manager_to_upstream_receiver_new) = - unbounded(); - let (upstream_to_channel_manager_sender_new, upstream_to_channel_manager_receiver_new) = - unbounded(); - let (channel_manager_to_jd_sender_new, channel_manager_to_jd_receiver_new) = unbounded(); - let (jd_to_channel_manager_sender_new, jd_to_channel_manager_receiver_new) = unbounded(); - - let (downstream_to_channel_manager_sender_new, downstream_to_channel_manager_receiver_new) = - unbounded(); - - // Create a fresh channel_manager with new channels - let channel_manager = match ChannelManager::new( - self.config.clone(), - channel_manager_to_upstream_sender_new.clone(), - upstream_to_channel_manager_receiver_new.clone(), - channel_manager_to_jd_sender_new.clone(), - jd_to_channel_manager_receiver_new.clone(), - channel_manager_to_tp_sender.clone(), - tp_to_channel_manager_receiver.clone(), - downstream_to_channel_manager_receiver_new.clone(), - encoded_outputs.clone(), - self.config.supported_extensions().to_vec(), - self.config.required_extensions().to_vec(), - mode.clone(), - ) - .await - { - Ok(channel_manager) => channel_manager, - Err(e) => { - error!(error = ?e, "Failed to initialize channel manager during fallback"); - self.cancellation_token.cancel(); - break; + RuntimeEvent::Fallback => { + let tp_ready_runtime = running.cleanup_for_fallback().await; + + running = match tp_ready_runtime.bootstrap_mining().await { + Ok(new_running) => new_running, + Err(bootstrap_err) => { + error!(?bootstrap_err.kind, "Failed to reconnect JDC"); + bootstrap_err.runtime.shutdown().await; + self.mark_stopped(); + return Err(bootstrap_err.kind); } }; - - channel_manager.clone() - .start( - self.cancellation_token.clone(), - fallback_coordinator.clone(), - task_manager.clone(), - miner_coinbase_outputs.clone(), - ) - .await; - - info!("Attempting to initialize Jd and upstream..."); - - match self - .initialize_jd( - &mut upstream_addresses, - channel_manager_to_upstream_receiver_new.clone(), - upstream_to_channel_manager_sender_new.clone(), - channel_manager_to_jd_receiver_new.clone(), - jd_to_channel_manager_sender_new.clone(), - self.cancellation_token.clone(), - fallback_coordinator.clone(), - mode.clone(), - task_manager.clone(), - ) - .await - { - Ok((upstream, job_declarator, user_identity)) => { - channel_manager.set_user_identity(user_identity); - - upstream - .start( - self.config.min_supported_version(), - self.config.max_supported_version(), - self.cancellation_token.clone(), - fallback_coordinator.clone(), - task_manager.clone(), - ) - .await; - - job_declarator - .start( - self.cancellation_token.clone(), - fallback_coordinator.clone(), - task_manager.clone(), - ) - .await; - - channel_manager - .upstream_state - .set(UpstreamState::NoChannel); - - _ = channel_manager.allocate_tokens(2).await; - } - Err(e) => { - tracing::error!("Failed to initialize upstream: {:?}", e); - channel_manager - .upstream_state - .set(UpstreamState::SoloMining); - mode.set_solo_mining(); - info!("Fallback to solo mining mode"); - } - }; - - // Reinitialize monitoring server if configured - #[cfg(feature = "monitoring")] - if let Some(monitoring_addr) = self.config.monitoring_address() { - info!("Reinitializing monitoring server on http://{monitoring_addr}"); - if let Err(e) = self.start_monitoring_tasks( - monitoring_addr, - channel_manager.clone(), - self.cancellation_token.clone(), - fallback_coordinator.clone(), - task_manager.clone(), - ) { - error!("Failed to reinitialize monitoring tasks: {e}"); - self.cancellation_token.cancel(); - break; - } - } - - task_manager.spawn({ - let config = self.config.clone(); - let cancellation_token = self.cancellation_token.clone(); - let task_manager = task_manager.clone(); - let fallback_coordinator = fallback_coordinator.clone(); - async move { - if let Err(e) = channel_manager - .start_downstream_server( - *config.authority_public_key(), - *config.authority_secret_key(), - config.cert_validity_sec(), - *config.listening_address(), - task_manager, - cancellation_token.clone(), - fallback_coordinator, - downstream_to_channel_manager_sender_new, - config.supported_extensions().to_vec(), - config.required_extensions().to_vec(), - ) - .await { - tracing::error!(?e, "Downstream server task exited with error"); - cancellation_token.cancel(); - } - } - }); } - _ = tokio::signal::ctrl_c() => { - info!("Ctrl+C received — initiating graceful shutdown..."); - self.cancellation_token.cancel(); - break; - } - } - } - - if let Some(bitcoin_core_sv2_cancellation_token) = bitcoin_core_sv2_cancellation_token { - bitcoin_core_sv2_cancellation_token.cancel(); - } - - if let Some(bitcoin_core_sv2_join_handle) = bitcoin_core_sv2_join_handle { - info!("Waiting for BitcoinCoreSv2TDP dedicated thread to shutdown..."); - match bitcoin_core_sv2_join_handle.join() { - Ok(_) => info!("BitcoinCoreSv2TDP dedicated thread shutdown complete."), - Err(e) => error!("BitcoinCoreSv2TDP dedicated thread error: {e:?}"), - } - } - - warn!( - "Graceful shutdown: waiting {} seconds for tasks to finish", - GRACEFUL_SHUTDOWN_TIMEOUT_SECONDS - ); - match tokio::time::timeout( - std::time::Duration::from_secs(GRACEFUL_SHUTDOWN_TIMEOUT_SECONDS), - task_manager.join_all(), - ) - .await - { - Ok(_) => { - info!("All tasks joined cleanly"); - } - Err(_) => { - warn!( - "Tasks did not finish within {} seconds, aborting", - GRACEFUL_SHUTDOWN_TIMEOUT_SECONDS - ); - task_manager.abort_all().await; - info!("Joining aborted tasks..."); - task_manager.join_all().await; - warn!("Forced shutdown complete"); } } - self.shutdown_notify.notify_waiters(); - self.is_alive.store(false, Ordering::Relaxed); - info!("JD Client shutdown complete."); - } - - #[cfg(feature = "monitoring")] - fn start_monitoring_tasks( - &self, - monitoring_addr: SocketAddr, - channel_manager: ChannelManager, - cancellation_token: CancellationToken, - fallback_coordinator: FallbackCoordinator, - task_manager: Arc, - ) -> Result<(), String> { - let refresh_interval = - Duration::from_secs(self.config.monitoring_cache_refresh_secs().unwrap_or(15)); - - let monitoring_server = stratum_apps::monitoring::MonitoringServer::new( - monitoring_addr, - Some(Arc::new(channel_manager.clone())), - Some(Arc::new(channel_manager.clone())), - refresh_interval, - ) - .map_err(|e| format!("failed to initialize monitoring server: {e}"))?; - - let cancellation_token_clone = cancellation_token.clone(); - let fallback_coordinator_token = fallback_coordinator.token(); - let shutdown_signal = async move { - tokio::select! { - _ = cancellation_token_clone.cancelled() => { - info!("Monitoring server: received shutdown signal."); - } - _ = fallback_coordinator_token.cancelled() => { - info!("Monitoring server: fallback triggered."); - } - } - }; - - let monitoring_fallback = fallback_coordinator.clone(); - task_manager.spawn({ - let cancellation_token = cancellation_token.clone(); - async move { - let fallback_handler = monitoring_fallback.register(); - - if let Err(e) = monitoring_server.run(shutdown_signal).await { - error!("Monitoring server error: {:?}", e); - cancellation_token.cancel(); - } - - fallback_handler.done(); - info!("Monitoring server task exited and signaled fallback coordinator"); - } - }); - - let telemetry_fallback = fallback_coordinator.clone(); - task_manager.spawn({ - async move { - let fallback_token = telemetry_fallback.token(); - let fallback_handler = telemetry_fallback.register(); - - channel_manager - .run_miner_telemetry_loop(refresh_interval, cancellation_token, fallback_token) - .await; - - fallback_handler.done(); - info!("JDC miner telemetry task exited and signaled fallback coordinator"); - } - }); - - Ok(()) } pub async fn shutdown(&self) { @@ -652,163 +115,6 @@ impl JobDeclaratorClient { self.cancellation_token.cancel(); notified.await; } - - /// Initializes an upstream pool + JD connection pair. - #[allow(clippy::too_many_arguments)] - pub async fn initialize_jd( - &self, - upstreams: &mut [UpstreamEntry], - channel_manager_to_upstream_receiver: Receiver, - upstream_to_channel_manager_sender: Sender, - channel_manager_to_jd_receiver: Receiver>, - jd_to_channel_manager_sender: Sender>, - cancellation_token: CancellationToken, - fallback_coordinator: FallbackCoordinator, - mode: JDMode, - task_manager: Arc, - ) -> Result<(Upstream, JobDeclarator, String), JDCErrorKind> { - const MAX_RETRIES: usize = 3; - let upstream_len = upstreams.len(); - for (i, upstream_entry) in upstreams.iter_mut().enumerate() { - info!( - "Trying upstream {} of {}: pool={}:{}, jds={}:{}", - i + 1, - upstream_len, - upstream_entry.pool_host, - upstream_entry.pool_port, - upstream_entry.jds_host, - upstream_entry.jds_port, - ); - - tokio::select! { - biased; - _ = cancellation_token.cancelled() => { - info!("Shutdown requested while waiting to initialize upstream, aborting retries"); - return Err(JDCErrorKind::CouldNotInitiateSystem); - } - _ = tokio::time::sleep(Duration::from_secs(1)) => {} - } - - if upstream_entry.tried_or_flagged { - info!( - "Upstream previously marked as malicious, skipping initial attempt warnings." - ); - continue; - } - - for attempt in 1..=MAX_RETRIES { - if cancellation_token.is_cancelled() { - info!( - "Shutdown requested before upstream connection attempt, aborting retries" - ); - return Err(JDCErrorKind::CouldNotInitiateSystem); - } - - info!("Connection attempt {}/{}...", attempt, MAX_RETRIES); - - match try_initialize_single( - upstream_entry, - upstream_to_channel_manager_sender.clone(), - channel_manager_to_upstream_receiver.clone(), - jd_to_channel_manager_sender.clone(), - channel_manager_to_jd_receiver.clone(), - cancellation_token.clone(), - fallback_coordinator.clone(), - mode.clone(), - task_manager.clone(), - &self.config, - ) - .await - { - Ok((upstream, jd)) => { - upstream_entry.tried_or_flagged = true; - return Ok((upstream, jd, upstream_entry.user_identity.clone())); - } - Err(e) => { - tracing::error!("Upstream and JDS connection terminated"); - - tokio::select! { - biased; - _ = cancellation_token.cancelled() => { - info!("Shutdown requested after upstream initialization failure, aborting retries"); - return Err(JDCErrorKind::CouldNotInitiateSystem); - } - _ = tokio::time::sleep(Duration::from_secs(1)) => {} - } - - warn!( - "Attempt {}/{} failed for pool={}:{}, jds={}:{}: {:?}", - attempt, - MAX_RETRIES, - upstream_entry.pool_host, - upstream_entry.pool_port, - upstream_entry.jds_host, - upstream_entry.jds_port, - e - ); - if attempt == MAX_RETRIES { - warn!( - "Max retries reached for pool={}:{}, jds={}:{}, moving to next upstream", - upstream_entry.pool_host, - upstream_entry.pool_port, - upstream_entry.jds_host, - upstream_entry.jds_port, - ); - } - } - } - } - upstream_entry.tried_or_flagged = true; - } - - tracing::error!("All upstreams failed after {} retries each", MAX_RETRIES); - Err(JDCErrorKind::CouldNotInitiateSystem) - } -} - -// Attempts to initialize a single upstream (pool + JDS pair). -#[allow(clippy::too_many_arguments)] -#[cfg_attr(not(test), hotpath::measure)] -async fn try_initialize_single( - upstream_entry: &UpstreamEntry, - upstream_to_channel_manager_sender: Sender, - channel_manager_to_upstream_receiver: Receiver, - jd_to_channel_manager_sender: Sender>, - channel_manager_to_jd_receiver: Receiver>, - cancellation_token: CancellationToken, - fallback_coordinator: FallbackCoordinator, - mode: JDMode, - task_manager: Arc, - config: &JobDeclaratorClientConfig, -) -> Result<(Upstream, JobDeclarator), JDCErrorKind> { - info!("Upstream connection in-progress at initialize single"); - let upstream = Upstream::new( - upstream_entry, - upstream_to_channel_manager_sender, - channel_manager_to_upstream_receiver, - cancellation_token.clone(), - fallback_coordinator.clone(), - task_manager.clone(), - config.required_extensions().to_vec(), - ) - .await - .map_err(|error| error.kind)?; - - info!("Upstream connection done at initialize single"); - - let job_declarator = JobDeclarator::new( - upstream_entry, - jd_to_channel_manager_sender, - channel_manager_to_jd_receiver, - cancellation_token, - fallback_coordinator, - mode, - task_manager.clone(), - ) - .await - .map_err(|error| error.kind)?; - - Ok((upstream, job_declarator)) } impl Drop for JobDeclaratorClient { diff --git a/miner-apps/jd-client/src/main.rs b/miner-apps/jd-client/src/main.rs index e77d70e39..7f63af23c 100644 --- a/miner-apps/jd-client/src/main.rs +++ b/miner-apps/jd-client/src/main.rs @@ -25,5 +25,7 @@ async fn inner_main() { }); init_logging(jdc_config.log_file()); - JobDeclaratorClient::new(jdc_config).start().await; + if JobDeclaratorClient::new(jdc_config).start().await.is_err() { + std::process::exit(1); + } }