diff --git a/.gitignore b/.gitignore index 3e075334..c2e034f8 100644 --- a/.gitignore +++ b/.gitignore @@ -1,3 +1,5 @@ Cargo.lock /target/ /.cache/ +# Bitcoin Core + SV2 template provider binaries downloaded by integration_tests_sv2 on first run +/mujina-miner/template-provider/ diff --git a/build.Containerfile b/build.Containerfile index 62e0f5dc..9ea16c75 100644 --- a/build.Containerfile +++ b/build.Containerfile @@ -12,6 +12,8 @@ FROM docker.io/library/rust:1.94-bookworm@sha256:b2fe2c0f26e0e1759752b6b2eb93b11 RUN rustup component add rustfmt clippy RUN apt-get update && apt-get install -y --no-install-recommends \ + capnproto \ + libcapnp-dev \ libudev-dev \ pkg-config \ && rm -rf /var/lib/apt/lists/* diff --git a/mujina-miner/Cargo.toml b/mujina-miner/Cargo.toml index 9e3939c3..aa8f97ab 100644 --- a/mujina-miner/Cargo.toml +++ b/mujina-miner/Cargo.toml @@ -45,6 +45,7 @@ rustix = { workspace = true } slotmap = { workspace = true } num-traits = { workspace = true } ruint = "1.17.0" +stratum-apps = { version = "0.4", default-features = false, features = ["network", "core", "config"] } [target.'cfg(target_os = "macos")'.dependencies] core-foundation = { workspace = true } @@ -73,6 +74,7 @@ skip-pty-tests = [] # Skip PTY-based serial tests that may hang in some environ [dev-dependencies] http = "1" http-body-util = "0.1" +integration_tests_sv2 = "0.2.0" serial_test = "3.3.1" test-case = { workspace = true } tokio = { workspace = true, features = ["test-util"] } diff --git a/mujina-miner/src/asic/bm13xx/thread.rs b/mujina-miner/src/asic/bm13xx/thread.rs index 6e874bdd..a07cd375 100644 --- a/mujina-miner/src/asic/bm13xx/thread.rs +++ b/mujina-miner/src/asic/bm13xx/thread.rs @@ -688,9 +688,10 @@ async fn bm13xx_thread_actor( chip_initialized = true; } - // Send initial job to chip + // Send initial job to chip. current_task is committed only + // after a successful send so a conversion failure doesn't + // leave the ntime ticker with a broken task. let chip_job_id = chip_jobs.insert(new_task.clone()); - let old_task = current_task.replace(new_task.clone()); match task_to_job_full(&new_task, chip_job_id) { Ok(job_data) => { if let Err(e) = chip_commands.send(protocol::Command::JobFull { job_data }).await { @@ -708,6 +709,7 @@ async fn bm13xx_thread_actor( continue; } } + let old_task = current_task.replace(new_task.clone()); { let mut s = status.write().unwrap(); @@ -741,9 +743,10 @@ async fn bm13xx_thread_actor( // Clear old jobs (old shares invalid) chip_jobs.clear(); - // Send initial job to chip + // Send initial job to chip. current_task is committed only + // after a successful send so a conversion failure doesn't + // leave the ntime ticker with a broken task. let chip_job_id = chip_jobs.insert(new_task.clone()); - let old_task = current_task.replace(new_task.clone()); match task_to_job_full(&new_task, chip_job_id) { Ok(job_data) => { if let Err(e) = chip_commands.send(protocol::Command::JobFull { job_data }).await { @@ -761,6 +764,7 @@ async fn bm13xx_thread_actor( continue; } } + let old_task = current_task.replace(new_task.clone()); { let mut s = status.write().unwrap(); @@ -1074,4 +1078,114 @@ mod tests { ); assert_eq!(result.merkle_root, *esp_miner_job::wire_tx::MERKLE_ROOT); } + + // Spawn a minimal actor backed by a no-op sink and an idle stream. + // Returns cmd_tx and the removal sender (must be kept alive for the actor + // to keep running; dropping it causes removal_rx.changed() to fire). + fn spawn_test_actor() -> ( + mpsc::Sender, + watch::Sender, + ) { + use futures::{sink, stream}; + let (cmd_tx, cmd_rx) = mpsc::channel(8); + let (evt_tx, _) = mpsc::channel(8); + let (removal_tx, removal_rx) = watch::channel(ThreadRemovalSignal::Running); + tokio::spawn(bm13xx_thread_actor( + cmd_rx, + evt_tx, + removal_rx, + Arc::new(RwLock::new(HashThreadStatus::default())), + stream::pending::>(), + sink::drain(), + BoardPeripherals { + asic_enable: None, + voltage_regulator: None, + }, + )); + (cmd_tx, removal_tx) + } + + // A Computed-merkle-root task with no EN2 value; task_to_job_full returns Err + // immediately on it, making it a convenient trigger for the conversion-failure path. + fn broken_computed_task() -> HashTask { + use crate::asic::bm13xx::test_data::esp_miner_job; + use crate::job_source::{ + Extranonce2Range, GeneralPurposeBits, JobTemplate, MerkleRootKind, MerkleRootTemplate, + VersionTemplate, + }; + let (share_tx, _) = mpsc::channel(1); + HashTask { + template: Arc::new(JobTemplate { + id: "broken".into(), + prev_blockhash: *esp_miner_job::wire_tx::PREV_BLOCKHASH, + version: VersionTemplate::new( + *esp_miner_job::wire_tx::VERSION, + GeneralPurposeBits::full(), + ) + .unwrap(), + bits: *esp_miner_job::wire_tx::NBITS, + share_target: crate::types::Difficulty::from(1_u64).to_target(), + time: 0, + merkle_root: MerkleRootKind::Computed(MerkleRootTemplate { + coinbase1: vec![], + extranonce1: vec![], + extranonce2_range: Extranonce2Range::new(4).unwrap(), + extranonce2_size: 4, + coinbase2: vec![], + merkle_branches: vec![], + }), + }), + en2_range: None, + en2: None, + share_target: crate::types::Difficulty::from(1_u64).to_target(), + ntime: 0, + share_tx, + } + } + + // Regression: a failed conversion must not commit current_task. + // GoIdle returns the current task; None means the broken task was never stored. + #[tokio::test(start_paused = true)] + async fn test_update_task_does_not_commit_on_conversion_failure() { + let (cmd_tx, _removal_tx) = spawn_test_actor(); + + let (tx, rx) = oneshot::channel(); + cmd_tx + .send(ThreadCommand::UpdateTask { + new_task: broken_computed_task(), + response_tx: tx, + }) + .await + .unwrap(); + rx.await.unwrap().unwrap_err(); + + let (tx, rx) = oneshot::channel(); + cmd_tx + .send(ThreadCommand::GoIdle { response_tx: tx }) + .await + .unwrap(); + assert!(rx.await.unwrap().unwrap().is_none()); + } + + #[tokio::test(start_paused = true)] + async fn test_replace_task_does_not_commit_on_conversion_failure() { + let (cmd_tx, _removal_tx) = spawn_test_actor(); + + let (tx, rx) = oneshot::channel(); + cmd_tx + .send(ThreadCommand::ReplaceTask { + new_task: broken_computed_task(), + response_tx: tx, + }) + .await + .unwrap(); + rx.await.unwrap().unwrap_err(); + + let (tx, rx) = oneshot::channel(); + cmd_tx + .send(ThreadCommand::GoIdle { response_tx: tx }) + .await + .unwrap(); + assert!(rx.await.unwrap().unwrap().is_none()); + } } diff --git a/mujina-miner/src/config.rs b/mujina-miner/src/config.rs index 43deb450..c686387a 100644 --- a/mujina-miner/src/config.rs +++ b/mujina-miner/src/config.rs @@ -6,6 +6,9 @@ use serde::{Deserialize, Serialize}; use std::path::{Path, PathBuf}; +use std::str::FromStr; +use stratum_apps::key_utils::Secp256k1PublicKey; +use thiserror::Error; /// Main configuration structure for the miner. #[derive(Debug, Clone, Deserialize, Serialize)] @@ -40,20 +43,177 @@ pub struct DaemonConfig { /// Pool connection configuration. #[derive(Debug, Clone, Deserialize, Serialize)] pub struct PoolConfig { - /// Pool URL (stratum+tcp://...) pub url: String, - - /// Worker name pub worker: String, - - /// Password (if required) pub password: Option, - - /// Priority (lower is higher priority) #[serde(default)] pub priority: u32, } +/// Error returned by [`PoolEndpoint::parse`]. +#[derive(Debug, Error, PartialEq, Eq)] +pub enum ParseEndpointError { + #[error( + "stratum2+tcp:// requires an authority public key in the path: stratum2+tcp://host:port/key" + )] + MissingAuthorityKey, + #[error("invalid authority public key: {0}")] + InvalidAuthorityKey(String), + #[error("missing port in endpoint (expected host:port)")] + MissingPort, + #[error("invalid port number '{0}'")] + InvalidPort(String), + #[error("empty host in endpoint")] + EmptyHost, + #[error("unsupported pool scheme '{0}'")] + UnsupportedScheme(String), +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize, Serialize)] +pub enum PoolProtocol { + StratumV1, + /// Stratum V1 over SSL/TLS (`stratum+ssl://`). + StratumV1Ssl, + /// Stratum V1 over SSL/TLS (`stratum+tls://`). + StratumV1Tls, + StratumV2, +} + +/// Parsed pool connection endpoint. +/// +/// Produced by [`PoolEndpoint::parse`] from the `url` field of [`PoolConfig`]. +/// +/// Recognised URL forms: +/// - `stratum2+tcp://host:port/authority_key` → [`PoolProtocol::StratumV2`] +/// - `stratum+tcp://host:port` → [`PoolProtocol::StratumV1`] +/// - `stratum+ssl://host:port` → [`PoolProtocol::StratumV1Ssl`] +/// - `stratum+tls://host:port` → [`PoolProtocol::StratumV1Tls`] +/// - `host:port` (bare) → [`PoolProtocol::StratumV1`] +#[derive(Debug, Clone)] +pub struct PoolEndpoint { + pub host: String, + pub port: u16, + pub protocol: PoolProtocol, + /// Present only for [`PoolProtocol::StratumV2`]; used in Noise_NX handshake. + /// + /// Private to enforce the invariant that V2 endpoints always carry a key + /// and V1 endpoints never do. Use [`PoolEndpoint::authority_pubkey`] to read it. + authority_pubkey: Option, +} + +impl PoolEndpoint { + /// Parse a pool URL string into a `PoolEndpoint`. + /// + /// Recognised URL forms: + /// - `stratum2+tcp://host:port/authority_key` → [`PoolProtocol::StratumV2`] + /// - `stratum+tcp://host:port` → [`PoolProtocol::StratumV1`] + /// - `stratum+ssl://host:port` → [`PoolProtocol::StratumV1Ssl`] + /// - `stratum+tls://host:port` → [`PoolProtocol::StratumV1Tls`] + /// - `host:port` (bare) → [`PoolProtocol::StratumV1`] + /// + /// Any other `scheme://` prefix returns [`ParseEndpointError::UnsupportedScheme`]. + pub fn parse(url: &str) -> Result { + // V2 is the only scheme that carries an authority key, so it parses on + // its own path. + if let Some(rest) = url.strip_prefix("stratum2+tcp://") { + return Self::parse_v2(rest); + } + + // The V1-family schemes differ only in their transport label. + const V1_SCHEMES: [(&str, PoolProtocol); 3] = [ + ("stratum+tcp://", PoolProtocol::StratumV1), + ("stratum+ssl://", PoolProtocol::StratumV1Ssl), + ("stratum+tls://", PoolProtocol::StratumV1Tls), + ]; + for (prefix, protocol) in V1_SCHEMES { + if let Some(rest) = url.strip_prefix(prefix) { + return Self::parse_v1(rest, protocol); + } + } + + if let Some(scheme_end) = url.find("://") { + return Err(ParseEndpointError::UnsupportedScheme( + url[..scheme_end].to_string(), + )); + } + + // Bare host:port defaults to plain V1. + Self::parse_v1(url, PoolProtocol::StratumV1) + } + + fn parse_v1(host_port: &str, protocol: PoolProtocol) -> Result { + let (host, port) = parse_host_port(host_port)?; + Ok(Self { + host, + port, + protocol, + authority_pubkey: None, + }) + } + + fn parse_v2(rest: &str) -> Result { + let (host_port, key_segment) = rest + .split_once('/') + .ok_or(ParseEndpointError::MissingAuthorityKey)?; + if key_segment.is_empty() { + return Err(ParseEndpointError::MissingAuthorityKey); + } + let (host, port) = parse_host_port(host_port)?; + let authority_pubkey = key_segment + .parse::() + .map_err(|e| ParseEndpointError::InvalidAuthorityKey(e.to_string()))?; + Ok(Self { + host, + port, + protocol: PoolProtocol::StratumV2, + authority_pubkey: Some(authority_pubkey), + }) + } + + /// Returns the authority public key for Stratum V2 endpoints (`None` for V1). + pub fn authority_pubkey(&self) -> Option { + self.authority_pubkey + } +} + +impl FromStr for PoolEndpoint { + type Err = ParseEndpointError; + + fn from_str(s: &str) -> Result { + Self::parse(s) + } +} + +fn parse_host_port(s: &str) -> Result<(String, u16), ParseEndpointError> { + let colon = s.rfind(':').ok_or(ParseEndpointError::MissingPort)?; + let host_part = &s[..colon]; + let port_str = &s[colon + 1..]; + + // Strip brackets from IPv6 addresses like [::1]. + let host = if host_part.starts_with('[') && host_part.ends_with(']') { + host_part[1..host_part.len() - 1].to_string() + } else { + host_part.to_string() + }; + + if host.is_empty() { + return Err(ParseEndpointError::EmptyHost); + } + + let port = port_str + .parse::() + .map_err(|_| ParseEndpointError::InvalidPort(port_str.to_string()))?; + + Ok((host, port)) +} + +impl PoolConfig { + /// Parse [`PoolConfig::url`] into a [`PoolEndpoint`]. + pub fn endpoint(&self) -> Result { + PoolEndpoint::parse(&self.url) + } +} + /// Hardware configuration. #[derive(Debug, Clone, Deserialize, Serialize)] pub struct HardwareConfig { @@ -99,3 +259,137 @@ impl Config { unimplemented!("Config loading not yet implemented") } } + +#[cfg(test)] +mod tests { + use super::*; + + // A known-good Base58Check-encoded SV2 authority public key (from stratum-apps test suite). + const VALID_KEY: &str = "9bDuixKmZqAJnrmP746n8zU1wyAQRrus7th9dxnkPg6RzQvCnan"; + // Same key with the last character swapped — invalid checksum. + const BAD_CHECKSUM_KEY: &str = "9bDuixKmZqAJnrmP746n8zU1wyAQRrus7th9dxnkPg6RzQvCnam"; + + #[test] + fn parse_stratum_v1_scheme() { + let ep = PoolEndpoint::parse("stratum+tcp://pool.example.com:3333").unwrap(); + assert_eq!(ep.host, "pool.example.com"); + assert_eq!(ep.port, 3333); + assert_eq!(ep.protocol, PoolProtocol::StratumV1); + assert!(ep.authority_pubkey().is_none()); + } + + #[test] + fn parse_stratum_v1_ssl_scheme() { + let ep = PoolEndpoint::parse("stratum+ssl://pool.example.com:3333").unwrap(); + assert_eq!(ep.host, "pool.example.com"); + assert_eq!(ep.port, 3333); + assert_eq!(ep.protocol, PoolProtocol::StratumV1Ssl); + assert!(ep.authority_pubkey().is_none()); + } + + #[test] + fn parse_stratum_v1_tls_scheme() { + let ep = PoolEndpoint::parse("stratum+tls://pool.example.com:3333").unwrap(); + assert_eq!(ep.host, "pool.example.com"); + assert_eq!(ep.port, 3333); + assert_eq!(ep.protocol, PoolProtocol::StratumV1Tls); + assert!(ep.authority_pubkey().is_none()); + } + + #[test] + fn parse_bare_host_port() { + let ep = PoolEndpoint::parse("pool.example.com:3333").unwrap(); + assert_eq!(ep.host, "pool.example.com"); + assert_eq!(ep.port, 3333); + assert_eq!(ep.protocol, PoolProtocol::StratumV1); + assert!(ep.authority_pubkey().is_none()); + } + + #[test] + fn parse_ipv4_bare() { + let ep = PoolEndpoint::parse("192.168.1.1:3333").unwrap(); + assert_eq!(ep.host, "192.168.1.1"); + assert_eq!(ep.port, 3333); + assert_eq!(ep.protocol, PoolProtocol::StratumV1); + } + + #[test] + fn parse_ipv6_bracketed() { + let ep = PoolEndpoint::parse("[::1]:3333").unwrap(); + assert_eq!(ep.host, "::1"); + assert_eq!(ep.port, 3333); + assert_eq!(ep.protocol, PoolProtocol::StratumV1); + } + + #[test] + fn parse_stratum_v2_valid() { + let url = format!("stratum2+tcp://pool.example.com:3336/{VALID_KEY}"); + let ep = PoolEndpoint::parse(&url).unwrap(); + assert_eq!(ep.host, "pool.example.com"); + assert_eq!(ep.port, 3336); + assert_eq!(ep.protocol, PoolProtocol::StratumV2); + assert!(ep.authority_pubkey().is_some()); + } + + #[test] + fn parse_v2_ipv6_with_key() { + let url = format!("stratum2+tcp://[::1]:3336/{VALID_KEY}"); + let ep = PoolEndpoint::parse(&url).unwrap(); + assert_eq!(ep.host, "::1"); + assert_eq!(ep.port, 3336); + assert_eq!(ep.protocol, PoolProtocol::StratumV2); + } + + #[test] + fn parse_v2_missing_key_no_slash() { + let err = PoolEndpoint::parse("stratum2+tcp://pool.example.com:3336").unwrap_err(); + assert_eq!(err, ParseEndpointError::MissingAuthorityKey); + } + + #[test] + fn parse_v2_missing_key_trailing_slash() { + let err = PoolEndpoint::parse("stratum2+tcp://pool.example.com:3336/").unwrap_err(); + assert_eq!(err, ParseEndpointError::MissingAuthorityKey); + } + + #[test] + fn parse_v2_malformed_key_bad_checksum() { + let url = format!("stratum2+tcp://pool.example.com:3336/{BAD_CHECKSUM_KEY}"); + let err = PoolEndpoint::parse(&url).unwrap_err(); + assert!(matches!(err, ParseEndpointError::InvalidAuthorityKey(_))); + } + + #[test] + fn parse_v2_malformed_key_garbage() { + let err = + PoolEndpoint::parse("stratum2+tcp://pool.example.com:3336/notavalidkey").unwrap_err(); + assert!(matches!(err, ParseEndpointError::InvalidAuthorityKey(_))); + } + + #[test] + fn parse_unsupported_scheme_errors() { + for url in [ + "tcp://pool.example.com:3333", + "stratum://pool.example.com:3333", + "http://pool.example.com:3333", + ] { + let err = PoolEndpoint::parse(url).unwrap_err(); + assert!( + matches!(err, ParseEndpointError::UnsupportedScheme(_)), + "expected UnsupportedScheme for {url}, got {err:?}" + ); + } + } + + #[test] + fn pool_config_endpoint_helper() { + let cfg = PoolConfig { + url: format!("stratum+tcp://pool.example.com:3333"), + worker: "worker".to_string(), + password: None, + priority: 0, + }; + let ep = cfg.endpoint().unwrap(); + assert_eq!(ep.protocol, PoolProtocol::StratumV1); + } +} diff --git a/mujina-miner/src/cpu_miner/hasher.rs b/mujina-miner/src/cpu_miner/hasher.rs index 152dfb5e..9ae91d0d 100644 --- a/mujina-miner/src/cpu_miner/hasher.rs +++ b/mujina-miner/src/cpu_miner/hasher.rs @@ -455,6 +455,7 @@ mod tests { merkle_root: MerkleRootKind::Computed(MerkleRootTemplate { coinbase1: block_881423::coinbase1_bytes().to_vec(), extranonce1: block_881423::extranonce1_bytes().to_vec(), + extranonce2_size: extranonce2_range.size, extranonce2_range, coinbase2: block_881423::coinbase2_bytes().to_vec(), merkle_branches: block_881423::MERKLE_BRANCHES.clone(), diff --git a/mujina-miner/src/daemon.rs b/mujina-miner/src/daemon.rs index ea916aa0..b3b23f83 100644 --- a/mujina-miner/src/daemon.rs +++ b/mujina-miner/src/daemon.rs @@ -5,6 +5,7 @@ use std::env; +use anyhow::Context as _; use tokio::signal::unix::{self, SignalKind}; use tokio::sync::{mpsc, watch}; use tokio_util::{sync::CancellationToken, task::TaskTracker}; @@ -14,16 +15,20 @@ use crate::tracing::prelude::*; use crate::{ api::{self, ApiConfig, commands::SchedulerCommand}, backplane::Backplane, + config::{PoolEndpoint, PoolProtocol}, cpu_miner::CpuMinerConfig, job_source::{ SourceCommand, SourceEvent, dummy::DummySource, forced_rate::{ForcedRateConfig, ForcedRateSource}, stratum_v1::StratumV1Source, + stratum_v2::StratumV2Source, }, scheduler::{self, SourceRegistration, ThreadRegistration}, stratum_v1::{PoolConfig as StratumPoolConfig, TcpConnector}, + stratum_v2::PoolConfig as Sv2PoolConfig, transport::{CpuDeviceInfo, TransportEvent, UsbTransport, cpu as cpu_transport}, + types::HashRate, }; /// The main daemon. @@ -119,97 +124,161 @@ impl Daemon { let (source_cmd_tx, source_cmd_rx) = mpsc::channel(10); if let Ok(pool_url) = env::var("MUJINA_POOL_URL") { - // Use Stratum v1 source - let pool_user = - env::var("MUJINA_POOL_USER").unwrap_or_else(|_| "mujina-testing".to_string()); - let pool_pass = env::var("MUJINA_POOL_PASS").unwrap_or_else(|_| "x".to_string()); - - let stratum_config = StratumPoolConfig { - url: pool_url.clone(), - username: pool_user, - password: pool_pass, - user_agent: "mujina-miner/0.1.0-alpha".to_string(), - }; - - // Optionally wrap with ForcedRateSource for testing - if let Some(forced_rate_config) = ForcedRateConfig::from_env() { - info!( - rate = %forced_rate_config.target_rate, - "Forced share rate wrapper enabled" - ); - - // Create inner channels (stratum <-> wrapper) - let (inner_event_tx, inner_event_rx) = mpsc::channel::(100); - let (inner_cmd_tx, inner_cmd_rx) = mpsc::channel::(10); - - let stratum_source = StratumV1Source::new( - stratum_config, - inner_cmd_rx, - inner_event_tx, - self.shutdown.clone(), - Box::new(TcpConnector::new(pool_url.clone())), - ); - let stratum_name = stratum_source.name(); - - // Spawn stratum source - self.tracker.spawn(async move { - if let Err(e) = stratum_source.run().await { - error!("Stratum v1 source error: {}", e); - } - }); - - // Create and spawn wrapper (uses outer channels from above) - let forced_rate = ForcedRateSource::new( - forced_rate_config, - inner_event_rx, - source_event_tx, - inner_cmd_tx, - source_cmd_rx, - self.shutdown.clone(), - ); - - source_reg_tx - .send(SourceRegistration { - name: format!("{} (forced-rate)", stratum_name), - url: Some(pool_url.clone()), - event_rx: source_event_rx, - command_tx: source_cmd_tx, - }) - .await?; - - self.tracker.spawn(async move { - if let Err(e) = forced_rate.run().await { - error!("Forced rate wrapper error: {}", e); - } - }); - } else { - // Direct stratum source (no wrapper) - let stratum_source = StratumV1Source::new( - stratum_config, - source_cmd_rx, - source_event_tx, - self.shutdown.clone(), - Box::new(TcpConnector::new(pool_url.clone())), - ); - - source_reg_tx - .send(SourceRegistration { - name: stratum_source.name(), - url: Some(pool_url), - event_rx: source_event_rx, - command_tx: source_cmd_tx, - }) - .await?; - - self.tracker.spawn(async move { - if let Err(e) = stratum_source.run().await { - error!("Stratum v1 source error: {}", e); + let endpoint = + PoolEndpoint::parse(&pool_url).context("Failed to parse MUJINA_POOL_URL")?; + + match endpoint.protocol { + PoolProtocol::StratumV2 => { + // Use Stratum V2 source. + // Safety: V2 endpoints always carry an authority public key; + // this invariant is enforced by PoolEndpoint::parse. + let authority_pubkey = endpoint + .authority_pubkey() + .expect("V2 endpoint always has authority pubkey"); + + let pool_user = env::var("MUJINA_POOL_USER") + .unwrap_or_else(|_| "mujina-testing".to_string()); + + info!( + host = %endpoint.host, + port = endpoint.port, + user = %pool_user, + "Using Stratum V2 source" + ); + + let sv2_config = Sv2PoolConfig::new( + endpoint.host.clone(), + endpoint.port, + authority_pubkey, + pool_user, + "mujina".to_string(), + "unknown".to_string(), + "mujina-miner/0.1.0-alpha".to_string(), + String::new(), + HashRate::default(), + ) + .context("Failed to build Stratum V2 pool config")?; + + let sv2_source = StratumV2Source::new( + sv2_config, + source_cmd_rx, + source_event_tx, + self.shutdown.clone(), + ); + + source_reg_tx + .send(SourceRegistration { + name: sv2_source.to_string(), + url: Some(pool_url), + event_rx: source_event_rx, + command_tx: source_cmd_tx, + }) + .await?; + + self.tracker.spawn(async move { + if let Err(e) = sv2_source.run().await { + error!("Stratum v2 source error: {}", e); + } + }); + } + PoolProtocol::StratumV1 => { + // Use Stratum v1 source + let pool_user = env::var("MUJINA_POOL_USER") + .unwrap_or_else(|_| "mujina-testing".to_string()); + let pool_pass = + env::var("MUJINA_POOL_PASS").unwrap_or_else(|_| "x".to_string()); + + let stratum_config = StratumPoolConfig { + url: pool_url.clone(), + username: pool_user, + password: pool_pass, + user_agent: "mujina-miner/0.1.0-alpha".to_string(), + }; + + // Optionally wrap with ForcedRateSource for testing + if let Some(forced_rate_config) = ForcedRateConfig::from_env() { + info!( + rate = %forced_rate_config.target_rate, + "Forced share rate wrapper enabled" + ); + + // Create inner channels (stratum <-> wrapper) + let (inner_event_tx, inner_event_rx) = mpsc::channel::(100); + let (inner_cmd_tx, inner_cmd_rx) = mpsc::channel::(10); + + let stratum_source = StratumV1Source::new( + stratum_config, + inner_cmd_rx, + inner_event_tx, + self.shutdown.clone(), + Box::new(TcpConnector::new(pool_url.clone())), + ); + let stratum_name = stratum_source.name(); + + // Spawn stratum source + self.tracker.spawn(async move { + if let Err(e) = stratum_source.run().await { + error!("Stratum v1 source error: {}", e); + } + }); + + // Create and spawn wrapper (uses outer channels from above) + let forced_rate = ForcedRateSource::new( + forced_rate_config, + inner_event_rx, + source_event_tx, + inner_cmd_tx, + source_cmd_rx, + self.shutdown.clone(), + ); + + source_reg_tx + .send(SourceRegistration { + name: format!("{} (forced-rate)", stratum_name), + url: Some(pool_url.clone()), + event_rx: source_event_rx, + command_tx: source_cmd_tx, + }) + .await?; + + self.tracker.spawn(async move { + if let Err(e) = forced_rate.run().await { + error!("Forced rate wrapper error: {}", e); + } + }); + } else { + // Direct stratum source (no wrapper) + let stratum_source = StratumV1Source::new( + stratum_config, + source_cmd_rx, + source_event_tx, + self.shutdown.clone(), + Box::new(TcpConnector::new(pool_url.clone())), + ); + + source_reg_tx + .send(SourceRegistration { + name: stratum_source.name(), + url: Some(pool_url), + event_rx: source_event_rx, + command_tx: source_cmd_tx, + }) + .await?; + + self.tracker.spawn(async move { + if let Err(e) = stratum_source.run().await { + error!("Stratum v1 source error: {}", e); + } + }); } - }); + } + PoolProtocol::StratumV1Ssl | PoolProtocol::StratumV1Tls => { + anyhow::bail!("Stratum V1 over SSL/TLS is not yet implemented"); + } } } else { // Use DummySource - info!("Using dummy job source (set MUJINA_POOL_URL to use Stratum v1)"); + info!("Using dummy job source (set MUJINA_POOL_URL to use Stratum v1 or v2)"); let dummy_source = DummySource::new( source_cmd_rx, diff --git a/mujina-miner/src/job_source/connection.rs b/mujina-miner/src/job_source/connection.rs new file mode 100644 index 00000000..5d516877 --- /dev/null +++ b/mujina-miner/src/job_source/connection.rs @@ -0,0 +1,98 @@ +//! Shared connection-lifecycle utilities for pool job sources. +//! +//! Both the Stratum V1 and V2 job sources share the same connection-lifecycle +//! primitives. Keeping them here ensures the behaviour stays in sync and +//! avoids copy-paste drift. + +use std::collections::hash_map::RandomState; +use std::hash::{BuildHasher, Hasher}; +use std::time::Duration; + +use tokio::sync::mpsc; +use tokio_util::sync::CancellationToken; + +/// Outcome of a single connection attempt, returned by each source's internal +/// `connect_and_run` method. +pub(super) enum ConnectOutcome { + /// Graceful shutdown requested via cancellation token. + Shutdown, + /// Connection lost; caller should retry after back-off. + Disconnected, + /// Unrecoverable error (e.g. auth failure, bad config); stop retrying. + Fatal(anyhow::Error), +} + +/// Jittered exponential back-off for reconnection timing. +/// +/// Starts at `initial` and doubles after each call to [`next_delay`], +/// capping at `max`. Each returned delay is scaled to [0.5, 1.0] of the +/// nominal value to spread out reconnection attempts and avoid +/// thundering-herd pile-ups against a recovering pool. +/// +/// [`next_delay`]: ExponentialBackoff::next_delay +pub(super) struct ExponentialBackoff { + current: Duration, + initial: Duration, + max: Duration, + // Per-process jitter seed. `RandomState` is seeded from OS randomness + // at construction, so different processes produce different jitter even + // when reconnecting at the same wall-clock instant — the same approach + // tokio uses internally for jittered timeouts. + jitter_state: RandomState, + jitter_step: u64, +} + +impl ExponentialBackoff { + pub(super) fn new(initial: Duration, max: Duration) -> Self { + Self { + current: initial, + initial, + max, + jitter_state: RandomState::new(), + jitter_step: 0, + } + } + + /// Return the next back-off delay (with jitter) and advance the state. + /// + /// The nominal delay (1 s, 2 s, 4 s, …) is scaled by a jitter factor in + /// [0.5, 1.0] to spread reconnection attempts across concurrent miners. + pub(super) fn next_delay(&mut self) -> Duration { + let nominal = self.current; + self.current = (self.current * 2).min(self.max); + + let mut hasher = self.jitter_state.build_hasher(); + hasher.write_u64(self.jitter_step); + self.jitter_step = self.jitter_step.wrapping_add(1); + let hash = hasher.finish(); + let jitter = 0.5 + (hash as f64 / u64::MAX as f64) * 0.5; + + nominal.mul_f64(jitter) + } + + /// Reset back-off to the initial delay (call after a stable connection). + pub(super) fn reset(&mut self) { + self.current = self.initial; + } +} + +/// Drain `command_rx` while sleeping for `delay`, returning `true` if shutdown +/// was requested before the sleep expired. +/// +/// Keeps the command channel drained so it does not back up during reconnect +/// waits. +pub(super) async fn backoff_wait( + delay: Duration, + command_rx: &mut mpsc::Receiver, + shutdown: &CancellationToken, +) -> bool { + let sleep = tokio::time::sleep(delay); + tokio::pin!(sleep); + loop { + tokio::select! { + _ = &mut sleep => return false, + Some(_) = command_rx.recv() => {}, + _ = shutdown.cancelled() => return true, + } + } +} diff --git a/mujina-miner/src/job_source/dummy.rs b/mujina-miner/src/job_source/dummy.rs index 26c030ac..5edcb285 100644 --- a/mujina-miner/src/job_source/dummy.rs +++ b/mujina-miner/src/job_source/dummy.rs @@ -95,6 +95,7 @@ impl DummySource { merkle_root: MerkleRootKind::Computed(MerkleRootTemplate { coinbase1: block_881423::coinbase1_bytes().to_vec(), extranonce1: block_881423::extranonce1_bytes().to_vec(), + extranonce2_size: extranonce2_range.size, extranonce2_range, coinbase2: block_881423::coinbase2_bytes().to_vec(), merkle_branches, diff --git a/mujina-miner/src/job_source/forced_rate.rs b/mujina-miner/src/job_source/forced_rate.rs index 480f046a..2a758f44 100644 --- a/mujina-miner/src/job_source/forced_rate.rs +++ b/mujina-miner/src/job_source/forced_rate.rs @@ -157,6 +157,8 @@ impl ForcedRateSource { SourceEvent::ReplaceJob(self.modify_job(job)) } SourceEvent::ClearJobs => SourceEvent::ClearJobs, + SourceEvent::SharesAccepted(n) => SourceEvent::SharesAccepted(n), + SourceEvent::SharesRejected => SourceEvent::SharesRejected, }; self.outer_event_tx.send(modified).await?; } diff --git a/mujina-miner/src/job_source/merkle.rs b/mujina-miner/src/job_source/merkle.rs index c8118722..3bcf26e7 100644 --- a/mujina-miner/src/job_source/merkle.rs +++ b/mujina-miner/src/job_source/merkle.rs @@ -52,9 +52,18 @@ pub struct MerkleRootTemplate { /// Extranonce2 range defining the available rolling space. /// /// The caller will create an iterator from this range to generate different - /// extranonce2 values for unique block headers. + /// extranonce2 values for unique block headers. The range's `size` field is + /// the counter width (1–8 bytes, capped by `u64`). pub extranonce2_range: Extranonce2Range, + /// Full extranonce2 allocation in bytes (≥ `extranonce2_range.size`). + /// Full extranonce2 byte length as negotiated with the pool (SV2 spec's + /// `extranonce_size`). When this exceeds the 8-byte u64 counter width, + /// the remaining bytes are zero-padded in both coinbase insertion and + /// `SubmitSharesExtended.extranonce`. For SV1/non-SV2 sources this always + /// equals `extranonce2_range.size`. + pub extranonce2_size: u8, + /// Second part of coinbase transaction (after extranonces). pub coinbase2: Vec, @@ -79,7 +88,14 @@ impl MerkleRootTemplate { let mut coinbase_bytes = Vec::new(); coinbase_bytes.extend_from_slice(&self.coinbase1); coinbase_bytes.extend_from_slice(&self.extranonce1); - extranonce2.extend_vec(&mut coinbase_bytes); + // Write counter bytes then zero-pad to the pool's full extranonce2_size. + // Per the SV2 spec, SubmitSharesExtended.extranonce MUST fill exactly + // `extranonce_size` bytes; when the pool allocates more than 8 bytes the + // u64 counter covers the first 1–8 bytes and the remainder is zeroed. + let mut en2_bytes = Vec::new(); + extranonce2.extend_vec(&mut en2_bytes); + en2_bytes.resize(self.extranonce2_size as usize, 0); + coinbase_bytes.extend_from_slice(&en2_bytes); coinbase_bytes.extend_from_slice(&self.coinbase2); // Parse and compute coinbase txid @@ -110,10 +126,12 @@ mod tests { let extranonce2 = *block_881423::EXTRANONCE2; // Construct a template from golden values + let en2_size = extranonce2.size(); let template = MerkleRootTemplate { coinbase1: block_881423::coinbase1_bytes().to_vec(), extranonce1: block_881423::extranonce1_bytes().to_vec(), - extranonce2_range: Extranonce2Range::new(extranonce2.size()).unwrap(), + extranonce2_range: Extranonce2Range::new(en2_size).unwrap(), + extranonce2_size: en2_size, coinbase2: block_881423::coinbase2_bytes().to_vec(), merkle_branches: block_881423::MERKLE_BRANCHES.clone(), }; diff --git a/mujina-miner/src/job_source/messages.rs b/mujina-miner/src/job_source/messages.rs index 1b9c296a..ea3bc4da 100644 --- a/mujina-miner/src/job_source/messages.rs +++ b/mujina-miner/src/job_source/messages.rs @@ -139,6 +139,20 @@ pub enum SourceEvent { /// Scheduler should cancel all work from this source and wait for new job. /// Used during pool disconnection or when awaiting new block. ClearJobs, + + /// Pool acknowledged acceptance of shares (SV2 only). + /// + /// `count` is the number of shares accepted, derived from the sequence + /// number range in `SubmitSharesSuccess`. SV1 has no per-batch ACK, so + /// this variant is only emitted by SV2 sources. + SharesAccepted(u32), + + /// Pool rejected a share (SV2 only). + /// + /// Emitted on `SubmitSharesError`. Each error message from the pool + /// identifies a single rejected sequence number, so this event always + /// represents exactly one rejection. + SharesRejected, } /// Commands to sources (pull, coordinator-initiated). diff --git a/mujina-miner/src/job_source/mod.rs b/mujina-miner/src/job_source/mod.rs index ea6f33ef..cb6558f7 100644 --- a/mujina-miner/src/job_source/mod.rs +++ b/mujina-miner/src/job_source/mod.rs @@ -51,6 +51,7 @@ //! scheduler enforces it. // Submodules +mod connection; pub mod dummy; mod extranonce2; pub mod forced_rate; @@ -58,6 +59,7 @@ pub(crate) mod job; mod merkle; mod messages; pub mod stratum_v1; +pub mod stratum_v2; pub mod test_blocks; mod version; diff --git a/mujina-miner/src/job_source/stratum_v1.rs b/mujina-miner/src/job_source/stratum_v1.rs index 7c0c9231..69d4958a 100644 --- a/mujina-miner/src/job_source/stratum_v1.rs +++ b/mujina-miner/src/job_source/stratum_v1.rs @@ -4,9 +4,7 @@ //! abstraction. It handles the conversion between Stratum protocol messages and //! the internal JobTemplate/Share types used by the scheduler. -use std::collections::hash_map::RandomState; use std::future; -use std::hash::{BuildHasher, Hasher}; use std::time::Duration; use anyhow::Result; @@ -20,6 +18,7 @@ use crate::stratum_v1::{ use crate::tracing::prelude::*; use crate::types::{Difficulty, HashRate, ShareRate}; +use super::connection::{ConnectOutcome, ExponentialBackoff}; use super::{ Extranonce2Range, GeneralPurposeBits, JobTemplate, MerkleRootKind, MerkleRootTemplate, Share, SourceCommand, SourceEvent, VersionTemplate, @@ -53,67 +52,6 @@ const SUGGEST_MIN_INTERVAL: Duration = Duration::from_secs(5); /// flapping pool that accepts and immediately drops. const STABLE_CONNECTION_THRESHOLD: Duration = Duration::from_secs(60); -/// Exponential backoff for reconnection timing. -/// -/// Starts at `initial` and doubles after each call to `next_delay()`, -/// capping at `max`. Each returned delay is jittered to [0.5, 1.0] of -/// the nominal value to avoid thundering-herd reconnections. -struct ExponentialBackoff { - current: Duration, - initial: Duration, - max: Duration, - // Per-process jitter seed. RandomState is seeded from OS randomness - // at construction, so different processes produce different jitter - // even when reconnecting at the same wall-clock instant. This is - // the same approach tokio uses internally for jittered timeouts. - jitter_state: RandomState, - jitter_step: u64, -} - -impl ExponentialBackoff { - fn new(initial: Duration, max: Duration) -> Self { - Self { - current: initial, - initial, - max, - jitter_state: RandomState::new(), - jitter_step: 0, - } - } - - /// Return the next backoff delay (with jitter) and advance the state. - /// - /// The nominal delay (1s, 2s, 4s, ...) is scaled by a jitter factor - /// in [0.5, 1.0] to spread out reconnection attempts across miners. - fn next_delay(&mut self) -> Duration { - let nominal = self.current; - self.current = (self.current * 2).min(self.max); - - let mut hasher = self.jitter_state.build_hasher(); - hasher.write_u64(self.jitter_step); - self.jitter_step = self.jitter_step.wrapping_add(1); - let hash = hasher.finish(); - let jitter = 0.5 + (hash as f64 / u64::MAX as f64) * 0.5; - - nominal.mul_f64(jitter) - } - - /// Reset backoff to the initial delay. - fn reset(&mut self) { - self.current = self.initial; - } -} - -/// Outcome of a single connection attempt. -enum ConnectOutcome { - /// Graceful shutdown requested. - Shutdown, - /// Connection lost; retry after backoff. - Disconnected, - /// Unrecoverable error (e.g. auth failure); stop retrying. - Fatal(anyhow::Error), -} - /// Stratum v1 job source. /// /// Wraps a StratumV1Client and bridges between the Stratum protocol and @@ -234,6 +172,7 @@ impl StratumV1Source { merkle_root: MerkleRootKind::Computed(MerkleRootTemplate { coinbase1: job.coinbase1, extranonce1: state.extranonce1.clone(), + extranonce2_size: extranonce2_range.size, extranonce2_range, coinbase2: job.coinbase2, merkle_branches: job.merkle_branches, diff --git a/mujina-miner/src/job_source/stratum_v2.rs b/mujina-miner/src/job_source/stratum_v2.rs new file mode 100644 index 00000000..5854e6a5 --- /dev/null +++ b/mujina-miner/src/job_source/stratum_v2.rs @@ -0,0 +1,1287 @@ +//! Stratum V2 Extended Channel job source. +//! +//! Bridges [`StratumV2Client`] events into the scheduler's job-source +//! abstraction. Converts `NewExtendedMiningJob`/`SetNewPrevHash` pairs into +//! [`JobTemplate`]s and manages the connection lifecycle with exponential +//! back-off. +//! +//! # Share Submission +//! +//! Shares received via [`SourceCommand::SubmitShare`] are forwarded to the pool +//! as `SubmitSharesExtended` messages. Each share gets a monotonically +//! increasing sequence number within the connection. Pending submits are +//! tracked in a `VecDeque`; they are drained on `SubmitSharesSuccess` and +//! removed individually on `SubmitSharesError`. +//! +//! ## Stale-Share Detection +//! +//! Job state is managed by an [`ExtendedChannel`] from `channels_sv2`. Active +//! and past jobs (valid within the current chain tip) are tracked by the +//! channel; stale jobs (superseded by a prior chain tip) are rejected. +//! Shares referencing an unknown or stale job are dropped without forwarding. +//! The version field in each `SubmitSharesExtended` is reconstructed by +//! zeroing the GP bits from the job's base version and OR-ing in the +//! hardware-rolled GP bits from the share. + +use std::collections::VecDeque; +use std::fmt; +use std::time::Duration; + +use anyhow::{Context as _, Result}; +use bitcoin::block::Version; +use bitcoin::hash_types::{BlockHash, TxMerkleNode}; +use bitcoin::hashes::Hash as _; +use bitcoin::pow::CompactTarget; +use stratum_apps::stratum_core::binary_sv2::B032; +use stratum_apps::stratum_core::channels_sv2::chain_tip::ChainTip; +use stratum_apps::stratum_core::channels_sv2::client::error::ExtendedChannelError; +use stratum_apps::stratum_core::channels_sv2::client::extended::{ExtendedChannel, ExtendedJob}; +use stratum_apps::stratum_core::channels_sv2::client::share_accounting::{ + ShareValidationError, ShareValidationResult, +}; +use stratum_apps::stratum_core::channels_sv2::extranonce_manager::ExtranoncePrefix; +use stratum_apps::stratum_core::mining_sv2::SubmitSharesExtended; +use tokio::sync::mpsc; +use tokio_util::sync::CancellationToken; + +use crate::stratum_v2::{ + ClientCommand, ClientEvent, ClientOutcome, PoolConfig, StratumV2Client, target_from_le_bytes, +}; +use crate::tracing::prelude::*; + +use super::connection::{ConnectOutcome, ExponentialBackoff, backoff_wait}; +use super::{ + Extranonce2Range, GeneralPurposeBits, JobTemplate, MerkleRootKind, MerkleRootTemplate, Share, + SourceCommand, SourceEvent, VersionTemplate, +}; + +/// Minimum reconnect back-off delay. +const INITIAL_BACKOFF: Duration = Duration::from_secs(1); +/// Maximum reconnect back-off delay. +const MAX_BACKOFF: Duration = Duration::from_secs(60); +/// Connections alive for at least this long reset the back-off on disconnect. +const STABLE_CONNECTION_THRESHOLD: Duration = Duration::from_secs(60); + +/// Per-connection channel state. +/// +/// Populated on `OpenExtendedMiningChannelSuccess` and replaced on each +/// reconnect. +struct SessionState { + channel_id: u32, + group_channel_id: u32, + /// The number of bytes in the extranonce (prefix + rollable). + extranonce_size: u8, + /// Rollable portion of the extranonce (capped by the u64 in `Extranonce2::value`). + extranonce_rollable_size: u8, + /// Monotonically increasing sequence number for `SubmitSharesExtended`. + next_seq: u32, + /// Sequence numbers of submitted-but-unacknowledged shares, in submission + /// order. Drained front-to-front on `SubmitSharesSuccess`, removed by + /// value on `SubmitSharesError`. + pending_submits: VecDeque, + /// Extended channel state: job lifecycle, target, extranonce prefix, and + /// share accounting. + channel: ExtendedChannel<'static>, +} + +impl SessionState { + /// Return true if `channel_id` targets this session. + /// + /// The `channel_id` field in [Mining Protocol > `NewExtendedMiningJob`][sv2-job] and + /// [Mining Protocol > `SetNewPrevHash`][sv2-prevhash] may address either the + /// individual channel or the group channel. Accept both; reject only messages + /// clearly meant for a different session. + /// + /// [sv2-job]: https://github.com/stratum-mining/sv2-spec/blob/main/05-Mining-Protocol.md#5316-newextendedminingjob-server---client + /// [sv2-prevhash]: https://github.com/stratum-mining/sv2-spec/blob/main/05-Mining-Protocol.md#5317-setnewprevhash-server---client-broadcast + fn accepts_channel_id(&self, channel_id: u32) -> bool { + channel_id == self.channel_id || channel_id == self.group_channel_id + } + + /// Drain all pending submit entries acknowledged by `last_seq`. + /// + /// `front` is considered ≤ `last_seq` when the forward distance from + /// `front` to `last_seq` in the circular u32 space is ≤ 2^31 − 1. + /// This is the "less than or equal to" relation from RFC 1982 § 3.2 + /// (), applied with + /// SERIAL_BITS = 32 so the threshold is `u32::MAX / 2` (= 2^31 − 1). + /// + /// The expression `last_seq.wrapping_sub(front) <= u32::MAX / 2` avoids + /// the half-window ambiguity of the `wrapping_sub as i32 <= 0` pattern, + /// which misclassifies a `front` exactly 2^31 steps ahead of `last_seq` + /// as already acknowledged (because `2^31 as i32 == i32::MIN < 0`). + fn acknowledge_up_to(&mut self, last_seq: u32) { + while let Some(&front) = self.pending_submits.front() { + if last_seq.wrapping_sub(front) <= u32::MAX / 2 { + self.pending_submits.pop_front(); + } else { + break; + } + } + } + + /// Remove the entry for `seq` from the pending queue (pool rejected it). + fn discard_pending(&mut self, seq: u32) { + self.pending_submits.retain(|&s| s != seq); + } +} + +/// Stratum V2 Extended Channel job source. +/// +/// Manages the connection lifecycle (DNS → TCP → Noise NX → session +/// negotiation) with exponential back-off and translates pool events into +/// [`SourceEvent`]s for the scheduler. +pub struct StratumV2Source { + config: PoolConfig, + event_tx: mpsc::Sender, + command_rx: mpsc::Receiver, + shutdown: CancellationToken, + session: Option, +} + +impl fmt::Display for StratumV2Source { + /// Human-readable pool name (`host:port`). + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "{}:{}", self.config.host(), self.config.port) + } +} + +impl StratumV2Source { + /// Create a new Stratum V2 job source. + pub fn new( + config: PoolConfig, + command_rx: mpsc::Receiver, + event_tx: mpsc::Sender, + shutdown: CancellationToken, + ) -> Self { + Self { + config, + event_tx, + command_rx, + shutdown, + session: None, + } + } + + /// Run the source (main entry point). + /// + /// Drains incoming commands until a positive hashrate is reported, then + /// enters the connect loop: attempt connection, serve events, and reconnect + /// with exponential back-off on disconnect. + pub async fn run(mut self) -> Result<()> { + info!( + host = %self.config.host(), + port = self.config.port, + "Waiting for hash threads before connecting" + ); + + // Wait for positive hashrate before opening a connection. + loop { + tokio::select! { + Some(cmd) = self.command_rx.recv() => { + match cmd { + SourceCommand::UpdateHashRate(rate) => { + if !rate.is_zero() { + break; + } + } + // No connection yet; drop shares silently. + SourceCommand::SubmitShare(_) => {} + } + } + _ = self.shutdown.cancelled() => return Ok(()), + } + } + + // Connect with automatic reconnection. + let mut backoff = ExponentialBackoff::new(INITIAL_BACKOFF, MAX_BACKOFF); + + loop { + self.session = None; + + info!(host = %self.config.host(), port = self.config.port, "Connecting to pool"); + + let connected_at = tokio::time::Instant::now(); + match self.connect_and_run().await { + ConnectOutcome::Shutdown => return Ok(()), + ConnectOutcome::Fatal(e) => { + error!(error = %e, "Fatal pool error, not reconnecting"); + return Err(e); + } + ConnectOutcome::Disconnected => { + // Invalidate stale work. + if let Err(e) = self.event_tx.send(SourceEvent::ClearJobs).await { + warn!(error = %e, "Failed to send ClearJobs"); + } + if connected_at.elapsed() >= STABLE_CONNECTION_THRESHOLD { + backoff.reset(); + } + let delay = backoff.next_delay(); + info!( + host = %self.config.host(), + delay_secs = delay.as_secs_f64(), + "Reconnecting after back-off" + ); + if self.backoff_wait(delay).await { + return Ok(()); + } + } + } + } + } + + /// Run a single connection attempt through its full lifecycle. + /// + /// Spawns the [`StratumV2Client`] task, runs the event loop until the + /// client exits (or shutdown is requested), and returns the outcome. + async fn connect_and_run(&mut self) -> ConnectOutcome { + let (client_event_tx, mut client_event_rx) = mpsc::channel::(100); + let (client_command_tx, client_command_rx) = mpsc::channel::(100); + + let client = StratumV2Client::new( + self.config.clone(), + client_event_tx, + client_command_rx, + self.shutdown.clone(), + ); + + let client_handle = tokio::spawn(async move { client.run().await }); + + loop { + tokio::select! { + event_opt = client_event_rx.recv() => { + match event_opt { + Some(event) => { + if let Err(e) = self.handle_client_event(event).await { + warn!(error = %e, "Error handling client event"); + } + } + None => { + // Client task exited; determine outcome below. + break; + } + } + } + + Some(cmd) = self.command_rx.recv() => { + match cmd { + SourceCommand::SubmitShare(share) => { + self.handle_share_submission(share, &client_command_tx).await; + } + SourceCommand::UpdateHashRate(_) => {} + } + } + + _ = self.shutdown.cancelled() => { + return ConnectOutcome::Shutdown; + } + } + } + + match client_handle.await { + Ok(Ok(ClientOutcome::Shutdown)) => ConnectOutcome::Shutdown, + Ok(Ok(ClientOutcome::ConnectionClosed | ClientOutcome::PoolRequestedReconnect)) => { + ConnectOutcome::Disconnected + } + Ok(Err(e)) => { + if e.is_fatal() { + ConnectOutcome::Fatal(e.into()) + } else { + warn!(error = %e, "Disconnected from pool"); + ConnectOutcome::Disconnected + } + } + Err(join_err) => { + ConnectOutcome::Fatal(anyhow::anyhow!("Client task panicked: {}", join_err)) + } + } + } + + /// Dispatch a single event from the [`StratumV2Client`]. + async fn handle_client_event(&mut self, event: ClientEvent) -> Result<()> { + match event { + ClientEvent::SetupConnectionSuccess { used_version, .. } => { + debug!(version = used_version, "SetupConnection accepted"); + } + + ClientEvent::OpenExtendedMiningChannelSuccess(msg) => { + let extranonce_prefix_bytes = msg.extranonce_prefix.inner_as_ref().to_vec(); + let extranonce_prefix = + ExtranoncePrefix::from_wire(extranonce_prefix_bytes.clone()).map_err(|e| { + anyhow::anyhow!( + "OpenExtendedMiningChannelSuccess: invalid extranonce prefix: {e}" + ) + })?; + + let extranonce_rollable_size = + Self::derive_extranonce_rollable_size(msg.extranonce_size) + .context("OpenExtendedMiningChannelSuccess: invalid extranonce params")?; + let extranonce_size = (msg.extranonce_size as usize).min(u8::MAX as usize) as u8; + let initial_target = target_from_le_bytes(msg.target.inner_as_ref()) + .context("OpenExtendedMiningChannelSuccess: invalid target")?; + + info!( + host = %self.config.host(), + channel_id = msg.channel_id, + extranonce_prefix = %hex::encode(&extranonce_prefix_bytes), + extranonce_rollable_size, + extranonce_size, + "Extended channel opened" + ); + + // version_rolling_allowed is set per-job by the pool on NewExtendedMiningJob. + // The channel-level flag here is used by validate_share() to enforce BIP320 + // compliance. Set true because Mujina performs version rolling in software. + let channel = ExtendedChannel::new( + msg.channel_id, + self.config.user_identity.as_utf8_or_hex(), + extranonce_prefix, + initial_target, + self.config.nominal_hash_rate.0 as f32, + true, + msg.extranonce_size, + ); + + self.session = Some(SessionState { + channel_id: msg.channel_id, + group_channel_id: msg.group_channel_id, + extranonce_rollable_size, + extranonce_size, + next_seq: 0, + pending_submits: VecDeque::new(), + channel, + }); + } + + ClientEvent::SetTarget(msg) => { + let target = target_from_le_bytes(msg.maximum_target.inner_as_ref()) + .context("SetTarget: invalid target")?; + if let Some(session) = &mut self.session { + if !session.accepts_channel_id(msg.channel_id) { + warn!( + expected = session.channel_id, + group = session.group_channel_id, + got = msg.channel_id, + "SetTarget channel_id mismatch; ignoring" + ); + return Ok(()); + } + debug!(channel_id = msg.channel_id, %target, "SetTarget"); + session.channel.set_target(target); + } + } + + ClientEvent::NewExtendedMiningJob(job) => { + let job_id = job.job_id; + let is_future = job.is_future(); + let Some(session) = &mut self.session else { + warn!(job_id, "Job arrived before channel opened; dropping"); + return Ok(()); + }; + if !session.accepts_channel_id(job.channel_id) { + warn!( + expected = session.channel_id, + group = session.group_channel_id, + got = job.channel_id, + job_id, + "NewExtendedMiningJob channel_id mismatch; ignoring" + ); + return Ok(()); + } + if is_future { + debug!(job_id, "Buffering future job"); + } + session + .channel + .on_new_extended_mining_job(job) + .map_err(|e| anyhow::anyhow!("NewExtendedMiningJob: {e:?}"))?; + if !is_future { + // Re-borrow session immutably — the mutable op is done. + let session = self.session.as_ref().unwrap(); + let Some(chain_tip) = session.channel.get_chain_tip() else { + warn!( + job_id, + "Non-future job arrived before SetNewPrevHash; dropping" + ); + return Ok(()); + }; + let active = session + .channel + .get_active_job() + .expect("active job must be set after non-future activation"); + let template = Self::build_job_template( + active, + chain_tip, + session.extranonce_rollable_size, + session.extranonce_size, + )?; + debug!(job_id, "Emitting ReplaceJob"); + self.event_tx + .send(SourceEvent::ReplaceJob(template)) + .await?; + } + } + + ClientEvent::SetNewPrevHash(msg) => { + let Some(session) = &mut self.session else { + warn!("SetNewPrevHash arrived before channel opened; dropping"); + return Ok(()); + }; + + if !session.accepts_channel_id(msg.channel_id) { + warn!( + expected = session.channel_id, + group = session.group_channel_id, + got = msg.channel_id, + "SetNewPrevHash channel_id mismatch; ignoring" + ); + return Ok(()); + } + + let prev_hash_hex = hex::encode(msg.prev_hash.inner_as_ref()); + let job_id = msg.job_id; + debug!( + job_id, + prev_hash = %prev_hash_hex, + nbits = format!("{:#010x}", msg.nbits), + "SetNewPrevHash" + ); + + match session.channel.on_set_new_prev_hash(msg) { + Ok(()) => { + // Re-borrow session immutably — the mutable op is done. + let session = self.session.as_ref().unwrap(); + let chain_tip = session.channel.get_chain_tip().unwrap(); + let active = session + .channel + .get_active_job() + .expect("active job must be set after on_set_new_prev_hash"); + let template = Self::build_job_template( + active, + chain_tip, + session.extranonce_rollable_size, + session.extranonce_size, + )?; + debug!(job_id, "Emitting ReplaceJob (future → active)"); + self.event_tx + .send(SourceEvent::ReplaceJob(template)) + .await?; + } + Err(ExtendedChannelError::JobIdNotFound) => { + // No future job was buffered for this prevhash — clear work + // and wait for the next NewExtendedMiningJob. + debug!( + job_id, + "SetNewPrevHash: no buffered future job; clearing work" + ); + if let Err(e) = self.event_tx.send(SourceEvent::ClearJobs).await { + warn!(error = %e, "Failed to send ClearJobs on SetNewPrevHash"); + } + } + Err(e) => return Err(anyhow::anyhow!("SetNewPrevHash: {e:?}")), + } + } + + // Pool-initiated reconnect signals. The client task will exit with + // PoolRequestedReconnect after emitting these events; no local flag needed. + ClientEvent::Reconnect(msg) => { + info!( + host = %self.config.host(), + new_host = msg.new_host.as_utf8_or_hex(), + new_port = msg.new_port, + "Pool requested reconnect" + ); + } + ClientEvent::ChannelEndpointChanged(msg) => { + info!( + channel_id = msg.channel_id, + "Channel endpoint changed; reconnecting" + ); + } + ClientEvent::CloseChannel(msg) => { + info!( + channel_id = msg.channel_id, + reason = msg.reason_code.as_utf8_or_hex(), + "Channel closed by pool; reconnecting" + ); + } + + ClientEvent::SubmitSharesSuccess(msg) => { + if let Some(session) = &mut self.session { + if !session.accepts_channel_id(msg.channel_id) { + warn!( + expected = session.channel_id, + group = session.group_channel_id, + got = msg.channel_id, + "SubmitSharesSuccess channel_id mismatch; ignoring" + ); + return Ok(()); + } + debug!( + channel_id = msg.channel_id, + last_seq = msg.last_sequence_number, + accepted = msg.new_submits_accepted_count, + "Pool accepted shares" + ); + session.acknowledge_up_to(msg.last_sequence_number); + session.channel.on_share_acknowledgement( + msg.new_submits_accepted_count, + msg.new_shares_sum as f64, + ); + if msg.new_submits_accepted_count > 0 { + self.event_tx + .send(SourceEvent::SharesAccepted(msg.new_submits_accepted_count)) + .await?; + } + } + } + + ClientEvent::SubmitSharesError(msg) => { + if let Some(session) = &mut self.session { + if !session.accepts_channel_id(msg.channel_id) { + warn!( + expected = session.channel_id, + group = session.group_channel_id, + got = msg.channel_id, + "SubmitSharesError channel_id mismatch; ignoring" + ); + return Ok(()); + } + warn!( + channel_id = msg.channel_id, + seq = msg.sequence_number, + reason = msg.error_code.as_utf8_or_hex(), + "Share rejected by pool" + ); + session.discard_pending(msg.sequence_number); + session + .channel + .on_share_rejection(msg.error_code.as_utf8_or_hex()); + self.event_tx.send(SourceEvent::SharesRejected).await?; + } + } + } + + Ok(()) + } + + /// Build and forward a share to the pool. + /// + /// Drops the share (with a trace/warn log) if: + /// - There is no active session (not yet connected or reconnecting). + /// - The job ID is not a valid `u32` (wrong protocol). + /// - Extranonce2 is missing from the share. + /// - Extranonce2 bytes cannot be encoded as B032 (> 32 bytes; should not + /// happen because `client.rs` rejects `extranonce_size > MAX_EXTRANONCE_SIZE (32)` + /// at channel open). + /// - Local share validation fails (stale job, below target, duplicate, or + /// version rolling violation); see [`ExtendedChannel::validate_share`]. + async fn handle_share_submission( + &mut self, + share: Share, + client_cmd_tx: &mpsc::Sender, + ) { + let Some(session) = &mut self.session else { + trace!(job_id = %share.job_id, "Share dropped: no active session"); + return; + }; + + let job_id: u32 = match share.job_id.parse() { + Ok(id) => id, + Err(_) => { + warn!( + job_id = %share.job_id, + "SV2 share has non-u32 job_id (wrong protocol?); dropping" + ); + return; + } + }; + + let Some(en2) = share.extranonce2 else { + warn!(job_id, "SV2 share missing extranonce2; dropping"); + return; + }; + + let mut en2_bytes: Vec = en2.into(); + // Zero-pad to the pool's full extranonce2 allocation when it exceeds + // the 8-byte counter width. This is due to a `u64` in `Extranonce2::value` + // limiting how many bytes extranonce could take of the coinbase script sig. + en2_bytes.resize(session.extranonce_size as usize, 0); + // Mining Protocol > SubmitSharesExtended: + // https://github.com/stratum-mining/sv2-spec/blob/main/05-Mining-Protocol.md#5312-submitsharesextended-client---server + // extranonce size MUST equal the negotiated extranonce_size; full coinbase: + // coinbase_tx_prefix + extranonce_prefix + extranonce + coinbase_tx_suffix. + let extranonce = match B032::try_from(en2_bytes) { + Ok(b) => b.into_static(), + Err(e) => { + warn!(job_id, error = ?e, "Failed to encode extranonce as B032; dropping"); + return; + } + }; + + let seq = session.next_seq; + session.next_seq = session.next_seq.wrapping_add(1); + session.pending_submits.push_back(seq); + + // VersionTemplate enforces that base_version has bits 13–28 clear; hardware + // only modifies those bits, so share.version is already the exact version + // used in the block header. + let sv2_share = SubmitSharesExtended { + channel_id: session.channel_id, + sequence_number: seq, + job_id, + nonce: share.nonce, + ntime: share.time, + version: share.version.to_consensus() as u32, + extranonce, + }; + + match session.channel.validate_share(sv2_share.clone()) { + Ok(ShareValidationResult::BlockFound(_)) => { + info!(job_id, seq, "Block candidate found"); + } + Ok(ShareValidationResult::Valid(_)) => {} + // Below-target shares still reach the pool — hardware/software filters at + // target before this point, so this is rare in practice. Pool feedback on + // these (e.g. difficulty-too-low after a SetTarget race) is more useful than + // a silent local drop. + Err(ShareValidationError::DoesNotMeetTarget) => {} + Err(e) => { + debug!(job_id, seq, error = ?e, "Share failed local validation; dropping"); + session.pending_submits.pop_back(); + return; + } + } + + trace!( + job_id, + seq, + nonce = format!("{:#010x}", share.nonce), + "Submitting share to pool" + ); + + if client_cmd_tx + .send(ClientCommand::SubmitShare(sv2_share)) + .await + .is_err() + { + warn!(seq, "Client disconnected before share could be forwarded"); + } + } + + /// Convert a [`NewExtendedMiningJob`] into a [`JobTemplate`]. + /// + /// The SV2 field mapping is: + /// + /// | SV2 field | `MerkleRootTemplate` field | + /// |------------------------|---------------------------| + /// | `coinbase_tx_prefix` | `coinbase1` | + /// | `extranonce_prefix` | `extranonce1` | + /// | `coinbase_tx_suffix` | `coinbase2` | + /// | `merkle_path` entries | `merkle_branches` | + fn build_job_template( + job: &ExtendedJob<'static>, + chain_tip: &ChainTip, + extranonce_rollable_size: u8, + extranonce_size: u8, + ) -> Result { + let (msg, extranonce_prefix, target) = job; + + let extranonce2_range = Extranonce2Range::new(extranonce_rollable_size) + .map_err(|e| anyhow::anyhow!("extranonce2 range: {e}"))?; + + // Version template. + // + // GP bits (13–28) are defined by BIP320 (https://github.com/bitcoin/bips/blob/master/bip-0320.mediawiki) + // for general-purpose version rolling. + // + // When version_rolling_allowed=true, the pool sets them to zero and + // the miner fills them — maps to GeneralPurposeBits::full(). + // + // When version_rolling_allowed=false, the version is fixed. + // VersionTemplate requires bits 13–28 clear in the base, so strip + // them. Compliant SV2 pools set them to zero in this case; if not, + // we warn and strip. + let raw_version = msg.version; + let gp_bits_mask = if msg.version_rolling_allowed { + GeneralPurposeBits::full() + } else { + if raw_version & 0x1fffe000 != 0 { + warn!( + job_id = msg.job_id, + version = format!("{:#010x}", raw_version), + "Non-rolling SV2 job has GP bits set; stripping them" + ); + } + GeneralPurposeBits::none() + }; + let base_version = Version::from_consensus((raw_version & !0x1fffe000) as i32); + let version_template = VersionTemplate::new(base_version, gp_bits_mask) + .map_err(|e| anyhow::anyhow!("VersionTemplate: {e}"))?; + + // Merkle path: Seq0255 → Vec. + // + // U256<'decoder> is a reference type (&[u8]), so iterating over + // inner_as_ref() yields &&[u8]; deref once to get &[u8]. + let merkle_branches: Vec = msg + .merkle_path + .inner_as_ref() + .iter() + .map(|u256| { + let bytes: [u8; 32] = (*u256) + .try_into() + .map_err(|_| anyhow::anyhow!("merkle branch is not 32 bytes"))?; + Ok(TxMerkleNode::from_byte_array(bytes)) + }) + .collect::>>()?; + + let prev_hash_bytes: [u8; 32] = chain_tip + .prev_hash() + .inner_as_ref() + .try_into() + .map_err(|_| anyhow::anyhow!("chain tip prev_hash is not 32 bytes"))?; + + Ok(JobTemplate { + id: msg.job_id.to_string(), + prev_blockhash: BlockHash::from_byte_array(prev_hash_bytes), + version: version_template, + bits: CompactTarget::from_consensus(chain_tip.nbits()), + share_target: *target, + // For immediate (non-future) jobs, min_ntime is set in the job message. + // For future jobs activated by SetNewPrevHash, min_ntime is None in the + // job and falls back to the value from the chain tip. + time: msg + .min_ntime + .clone() + .into_inner() + .unwrap_or_else(|| chain_tip.min_ntime()), + merkle_root: MerkleRootKind::Computed(MerkleRootTemplate { + coinbase1: msg.coinbase_tx_prefix.inner_as_ref().to_vec(), + extranonce1: extranonce_prefix.clone(), + extranonce2_range, + extranonce2_size: extranonce_size, + coinbase2: msg.coinbase_tx_suffix.inner_as_ref().to_vec(), + merkle_branches, + }), + }) + } + + /// Derive the rollable portion size from the pool's `extranonce_size`. + /// + /// `extranonce_size` is the miner's total extranonce allocation per the SV2 + /// spec (`OpenExtendedMiningChannelSuccess.extranonce_size`). It does NOT + /// include the pool's fixed `extranonce_prefix`. + /// + /// Because [`Extranonce2`] stores the counter as a `u64`, the rollable size + /// is capped at 8. When the pool allocates more than 8 bytes the miner wraps (restart from zero) + /// the first 8 and zero-pads the remainder in the coinbase and in + /// `SubmitSharesExtended.extranonce`. + fn derive_extranonce_rollable_size(extranonce_size: u16) -> Result { + if extranonce_size == 0 { + return Err(anyhow::anyhow!( + "extranonce_size 0 out of range (must be ≥ 1 byte)" + )); + } + Ok((extranonce_size as usize).min(8) as u8) + } + + /// Drain commands during back-off sleep. + /// + /// Returns `true` if shutdown was requested during the wait. + async fn backoff_wait(&mut self, delay: Duration) -> bool { + backoff_wait(delay, &mut self.command_rx, &self.shutdown).await + } +} + +// --------------------------------------------------------------------------- +// Tests +// --------------------------------------------------------------------------- + +#[cfg(test)] +mod tests { + use super::super::Extranonce2; + use super::*; + use bitcoin::pow::Target; + + // ---- derive_extranonce_rollable_size ---- + + #[test] + fn extranonce_rollable_size_typical() { + // Pool's extranonce_size = 4 bytes → counter width 4. + assert_eq!( + StratumV2Source::derive_extranonce_rollable_size(4).unwrap(), + 4 + ); + } + + #[test] + fn extranonce_rollable_size_valid_boundary() { + // Pool's extranonce_size = 8 bytes → counter width 8 (maximum). + assert_eq!( + StratumV2Source::derive_extranonce_rollable_size(8).unwrap(), + 8 + ); + + // Pool's extranonce_size = 1 byte → counter width 1 (minimum). + assert_eq!( + StratumV2Source::derive_extranonce_rollable_size(1).unwrap(), + 1 + ); + } + + #[test] + fn extranonce_rollable_size_zero_rejected() { + let err = StratumV2Source::derive_extranonce_rollable_size(0).unwrap_err(); + assert!(err.to_string().contains("out of range")); + } + + #[test] + fn extranonce_rollable_size_capped_when_pool_allocates_more_than_8() { + // Pool's extranonce_size = 16 bytes; counter is capped at 8, remainder zero-padded. + assert_eq!( + StratumV2Source::derive_extranonce_rollable_size(16).unwrap(), + 8 + ); + } + + // ---- target_from_le_bytes ---- + + #[test] + fn target_from_le_bytes_roundtrip() { + let target = Target::MAX; + let le_bytes = target.to_le_bytes(); + let recovered = target_from_le_bytes(&le_bytes).unwrap(); + assert_eq!(recovered, target); + } + + #[test] + fn target_from_le_bytes_wrong_length() { + let err = target_from_le_bytes(&[0u8; 16]).unwrap_err(); + assert!(err.to_string().contains("32-byte")); + } + + // ---- ExponentialBackoff ---- + + #[test] + fn backoff_doubles_each_step() { + let mut b = ExponentialBackoff::new(Duration::from_secs(1), Duration::from_secs(60)); + let d1 = b.next_delay(); + let d2 = b.next_delay(); + let d3 = b.next_delay(); + + // Nominal: 1 s, 2 s, 4 s. Jitter in [0.5, 1.0]. + assert!(d1 >= Duration::from_millis(500) && d1 < Duration::from_secs(1)); + assert!(d2 >= Duration::from_secs(1) && d2 < Duration::from_secs(2)); + assert!(d3 >= Duration::from_secs(2) && d3 < Duration::from_secs(4)); + } + + #[test] + fn backoff_caps_at_max() { + let mut b = ExponentialBackoff::new(Duration::from_secs(32), Duration::from_secs(60)); + b.next_delay(); // 32 s nominal + let d = b.next_delay(); // capped at 60 s → jittered to [30, 60) + assert!(d >= Duration::from_secs(30) && d < Duration::from_secs(60)); + } + + #[test] + fn backoff_reset_restores_initial() { + let mut b = ExponentialBackoff::new(Duration::from_secs(1), Duration::from_secs(60)); + b.next_delay(); + b.next_delay(); + b.reset(); + let d = b.next_delay(); + assert!(d >= Duration::from_millis(500) && d < Duration::from_secs(1)); + } + + fn make_session() -> SessionState { + let extranonce_prefix = ExtranoncePrefix::from_wire(vec![0xde, 0xad]).unwrap(); + let channel = ExtendedChannel::new( + 1, + "test".to_string(), + extranonce_prefix, + Target::MAX, + 1.0, + false, + 4, + ); + SessionState { + channel_id: 1, + group_channel_id: 0, + extranonce_rollable_size: 4, + extranonce_size: 4, + next_seq: 0, + pending_submits: VecDeque::new(), + channel, + } + } + + /// Contract: SubmitSharesSuccess removes all pending entries up to and + /// including last_sequence_number; later entries are untouched. + #[test] + fn success_drains_acknowledged_sequences() { + let mut session = make_session(); + session.pending_submits.extend([0, 1, 2, 3, 4]); + + session.acknowledge_up_to(2); + + assert_eq!(session.pending_submits, VecDeque::from([3, 4])); + } + + /// Contract: acknowledge_up_to is correct when next_seq has wrapped + /// around u32::MAX (entries near MAX must compare as before entries + /// near 0 when last_seq is near 0). + #[test] + fn success_handles_sequence_wraparound() { + let mut session = make_session(); + session + .pending_submits + .extend([u32::MAX - 1, u32::MAX, 0, 1]); + + session.acknowledge_up_to(0); + + assert_eq!(session.pending_submits, VecDeque::from([1])); + } + + /// Contract: an entry exactly 2^31 steps ahead of last_seq is NOT + /// acknowledged — it lies on the ambiguous boundary of the RFC 1982 + /// half-window and must be treated as a future (un-acked) entry. + #[test] + fn success_does_not_drain_half_window_boundary() { + let mut session = make_session(); + // 2^31 = u32::MAX / 2 + 1 steps ahead of last_seq = 0. + let half_window_plus_one = (u32::MAX / 2).wrapping_add(1); + session.pending_submits.extend([0, 1, half_window_plus_one]); + + session.acknowledge_up_to(0); + + // Only seq 0 (== last_seq) is drained; 1 and the boundary entry stay. + assert_eq!( + session.pending_submits, + VecDeque::from([1, half_window_plus_one]) + ); + } + + /// Contract: SubmitSharesError removes only the rejected entry; + /// all other pending entries are untouched. + #[test] + fn error_removes_rejected_sequence() { + let mut session = make_session(); + session.pending_submits.extend([0, 1, 2, 3]); + + session.discard_pending(2); + + assert_eq!(session.pending_submits, VecDeque::from([0, 1, 3])); + } + + #[test] + fn extranonce2_encoding_little_endian() { + let en2 = Extranonce2::new(0x0102_0304, 4).unwrap(); + let bytes: Vec = en2.into(); + assert_eq!(bytes, vec![0x04, 0x03, 0x02, 0x01]); + } + + #[test] + fn extranonce2_encoding_single_byte() { + let en2 = Extranonce2::new(0xab, 1).unwrap(); + let bytes: Vec = en2.into(); + assert_eq!(bytes, vec![0xab]); + } + + // ---- build_job_template ---- + + use bitcoin::{hash_types::BlockHash, pow::CompactTarget}; + use stratum_apps::stratum_core::{ + binary_sv2::{B064K, Seq0255, Sv2Option, U256}, + mining_sv2::{NewExtendedMiningJob, SetNewPrevHash as SetNewPrevHashMp}, + }; + + /// Construct a `NewExtendedMiningJob<'static>` with the given parameters. + fn make_job( + job_id: u32, + version: u32, + version_rolling_allowed: bool, + ntime: Option, // None → future job + coinbase_prefix: Vec, + coinbase_suffix: Vec, + merkle_hashes: Vec<[u8; 32]>, + ) -> NewExtendedMiningJob<'static> { + let coinbase_tx_prefix = B064K::try_from(coinbase_prefix) + .expect("valid coinbase prefix") + .into_static(); + let coinbase_tx_suffix = B064K::try_from(coinbase_suffix) + .expect("valid coinbase suffix") + .into_static(); + let merkle_path_items: Vec> = merkle_hashes + .into_iter() + .map(|h| { + U256::try_from(h.to_vec()) + .expect("valid 32-byte hash") + .into_static() + }) + .collect(); + let merkle_path: Seq0255<'static, U256<'static>> = Seq0255::new(merkle_path_items) + .expect("≤255 items") + .into_static(); + NewExtendedMiningJob { + channel_id: 1, + job_id, + min_ntime: Sv2Option::new(ntime), + version, + version_rolling_allowed, + merkle_path, + coinbase_tx_prefix, + coinbase_tx_suffix, + } + } + + /// Construct a [`ChainTip`] with a zeroed prev_hash and the given + /// `min_ntime` / `nbits` values. + fn make_chain_tip(min_ntime: u32, nbits: u32) -> ChainTip { + ChainTip::from(SetNewPrevHashMp { + channel_id: 0, + job_id: 0, + prev_hash: [0u8; 32].into(), + nbits, + min_ntime, + }) + } + + /// Contract: coinbase_tx_prefix → coinbase1, extranonce_prefix → + /// extranonce1, coinbase_tx_suffix → coinbase2, merkle_path → + /// merkle_branches. job_id, prev_hash, nbits, and share_target are also + /// forwarded without modification. + #[test] + fn build_job_template_maps_all_fields() { + let prefix = vec![0xAA, 0xBB, 0xCC]; + let suffix = vec![0xDD, 0xEE, 0xFF]; + let job = make_job( + 42, + 0x2000_0000, + false, + Some(0x0102_0304), // immediate job + prefix.clone(), + suffix.clone(), + vec![], + ); + let chain_tip = make_chain_tip(0x0102_0304, 0x1d00_ffff); + let tmpl = StratumV2Source::build_job_template( + &(job, vec![0xde, 0xad], Target::MAX), + &chain_tip, + 4, + 4, + ) + .unwrap(); + + assert_eq!(tmpl.id, "42"); + assert_eq!(tmpl.prev_blockhash, BlockHash::all_zeros()); + assert_eq!(tmpl.bits, CompactTarget::from_consensus(0x1d00_ffff)); + assert_eq!(tmpl.share_target, Target::MAX); + + let MerkleRootKind::Computed(mrt) = tmpl.merkle_root else { + panic!("expected MerkleRootKind::Computed"); + }; + assert_eq!(mrt.coinbase1, prefix); + assert_eq!(mrt.extranonce1, vec![0xde, 0xad]); + assert_eq!(mrt.coinbase2, suffix); + assert!(mrt.merkle_branches.is_empty()); + } + + /// Contract: for an immediate job, `template.time` comes from the job's + /// own `min_ntime`. For a future job, it falls back to + /// `chain_tip.min_ntime()`. + #[test] + fn build_job_template_ntime_from_job_or_prevhash() { + // Immediate job: job's ntime wins. + let job_ntime = 0xAABB_CCDD; + let prevhash_ntime = 0x1111_2222; + let immediate = make_job( + 1, + 0x2000_0000, + false, + Some(job_ntime), + vec![], + vec![], + vec![], + ); + let chain_tip = make_chain_tip(prevhash_ntime, 0x1d00_ffff); + let tmpl = StratumV2Source::build_job_template( + &(immediate, vec![0xde, 0xad], Target::MAX), + &chain_tip, + 4, + 4, + ) + .unwrap(); + assert_eq!( + tmpl.time, job_ntime, + "immediate job must use its own min_ntime" + ); + + // Future job: chain tip ntime is the fallback. + let future = make_job(2, 0x2000_0000, false, None, vec![], vec![], vec![]); + let tmpl = StratumV2Source::build_job_template( + &(future, vec![0xde, 0xad], Target::MAX), + &chain_tip, + 4, + 4, + ) + .unwrap(); + assert_eq!( + tmpl.time, prevhash_ntime, + "future job must fall back to chain_tip.min_ntime()" + ); + } + + /// Contract: when `version_rolling_allowed = false`, the GP-bits mask in + /// the resulting [`VersionTemplate`] is `GeneralPurposeBits::none()`. + /// When `version_rolling_allowed = true`, the mask is + /// `GeneralPurposeBits::full()`. + #[test] + fn build_job_template_version_rolling_flag() { + let chain_tip = make_chain_tip(0, 0x1d00_ffff); + + let fixed = make_job(1, 0x2000_0000, false, Some(0), vec![], vec![], vec![]); + let tmpl = StratumV2Source::build_job_template( + &(fixed, vec![0xde, 0xad], Target::MAX), + &chain_tip, + 4, + 4, + ) + .unwrap(); + assert_eq!( + tmpl.version.gp_bits_mask(), + GeneralPurposeBits::none(), + "fixed-version job must have no GP bits" + ); + + let rolling = make_job(2, 0x2000_0000, true, Some(0), vec![], vec![], vec![]); + let tmpl = StratumV2Source::build_job_template( + &(rolling, vec![0xde, 0xad], Target::MAX), + &chain_tip, + 4, + 4, + ) + .unwrap(); + assert_eq!( + tmpl.version.gp_bits_mask(), + GeneralPurposeBits::full(), + "version-rolling job must expose all GP bits" + ); + } + + /// Contract: each 32-byte entry in `merkle_path` becomes one + /// [`TxMerkleNode`] in `merkle_branches`, preserving order. + #[test] + fn build_job_template_merkle_path_forwarded() { + use bitcoin::hash_types::TxMerkleNode; + use bitcoin::hashes::Hash as _; + + let hash_a = [0xABu8; 32]; + let hash_b = [0x12u8; 32]; + let job = make_job( + 1, + 0x2000_0000, + false, + Some(0), + vec![], + vec![], + vec![hash_a, hash_b], + ); + let chain_tip = make_chain_tip(0, 0x1d00_ffff); + let tmpl = StratumV2Source::build_job_template( + &(job, vec![0xde, 0xad], Target::MAX), + &chain_tip, + 4, + 4, + ) + .unwrap(); + let MerkleRootKind::Computed(mrt) = tmpl.merkle_root else { + panic!("expected MerkleRootKind::Computed"); + }; + assert_eq!(mrt.merkle_branches.len(), 2); + assert_eq!( + mrt.merkle_branches[0], + TxMerkleNode::from_byte_array(hash_a) + ); + assert_eq!( + mrt.merkle_branches[1], + TxMerkleNode::from_byte_array(hash_b) + ); + } + + // ---- accepts_channel_id ---- + + /// Contract: a session accepts messages addressed to its individual + /// channel_id, its group_channel_id, and rejects anything else. + #[test] + fn session_accepts_individual_and_group_channel_id() { + let session = SessionState { + channel_id: 2, + group_channel_id: 1, + ..make_session() + }; + + assert!( + session.accepts_channel_id(2), + "must accept individual channel_id" + ); + assert!( + session.accepts_channel_id(1), + "must accept group_channel_id" + ); + assert!(!session.accepts_channel_id(3), "must reject unrelated id"); + assert!(!session.accepts_channel_id(0), "must reject zero id"); + } + + // ---- extranonce_size propagation ---- + + /// Contract: extranonce_size is forwarded from the session into the + /// MerkleRootTemplate so compute_merkle_root can zero-pad en2 correctly + /// when the pool's extranonce_size exceeds 8 bytes. + #[test] + fn build_job_template_extranonce_size_propagated() { + let chain_tip = make_chain_tip(0, 0x1d00_ffff); + let job = make_job(1, 0x2000_0000, false, Some(0), vec![], vec![], vec![]); + let tmpl = StratumV2Source::build_job_template( + &(job, vec![0xde, 0xad], Target::MAX), + &chain_tip, + 8, + 16, + ) + .unwrap(); + + let MerkleRootKind::Computed(mrt) = tmpl.merkle_root else { + panic!("expected MerkleRootKind::Computed"); + }; + assert_eq!( + mrt.extranonce2_size, 16, + "extranonce_size must match the pool extranonce_size" + ); + assert_eq!( + mrt.extranonce2_range.size, 8, + "counter width must remain capped at 8" + ); + } + + /// Contract: when extranonce_size == extranonce_rollable_size (the typical case), + /// extranonce2_size equals extranonce2_range.size in the resulting template. + #[test] + fn build_job_template_extranonce_size_equals_rollable_when_not_padded() { + let chain_tip = make_chain_tip(0, 0x1d00_ffff); + let job = make_job(1, 0x2000_0000, false, Some(0), vec![], vec![], vec![]); + let tmpl = StratumV2Source::build_job_template( + &(job, vec![0xde, 0xad], Target::MAX), + &chain_tip, + 4, + 4, + ) + .unwrap(); + + let MerkleRootKind::Computed(mrt) = tmpl.merkle_root else { + panic!("expected MerkleRootKind::Computed"); + }; + assert_eq!( + mrt.extranonce2_size, mrt.extranonce2_range.size, + "extranonce_size and rollable size must match when the pool does not over-allocate" + ); + } +} diff --git a/mujina-miner/src/lib.rs b/mujina-miner/src/lib.rs index 05285d0b..cf7b8f38 100644 --- a/mujina-miner/src/lib.rs +++ b/mujina-miner/src/lib.rs @@ -13,6 +13,7 @@ pub mod mgmt_protocol; pub mod peripheral; pub mod scheduler; pub mod stratum_v1; +pub mod stratum_v2; pub mod tracing; pub mod transport; pub mod types; diff --git a/mujina-miner/src/scheduler.rs b/mujina-miner/src/scheduler.rs index e9f44bfa..c69b39a7 100644 --- a/mujina-miner/src/scheduler.rs +++ b/mujina-miner/src/scheduler.rs @@ -902,6 +902,20 @@ impl Scheduler { SourceEvent::ClearJobs => { self.handle_clear_jobs(source_id, &mut share_channels); } + + SourceEvent::SharesAccepted(count) => { + debug!( + source = %source_name, + count, + "Shares accepted by pool" + ); + self.stats.shares_accepted += u64::from(count); + } + + SourceEvent::SharesRejected => { + debug!(source = %source_name, "Share rejected by pool"); + self.stats.shares_rejected += 1; + } } } @@ -1137,6 +1151,8 @@ impl StartupGate { struct MiningStats { start_time: std::time::Instant, shares_submitted: u64, + shares_accepted: u64, + shares_rejected: u64, } impl Default for MiningStats { @@ -1144,6 +1160,8 @@ impl Default for MiningStats { Self { start_time: std::time::Instant::now(), shares_submitted: 0, + shares_accepted: 0, + shares_rejected: 0, } } } @@ -1161,7 +1179,9 @@ impl MiningStats { info!( uptime = %format_duration(elapsed.as_secs()), hashrate = %hashrate_str, - shares = self.shares_submitted, + shares_submitted = self.shares_submitted, + shares_accepted = self.shares_accepted, + shares_rejected = self.shares_rejected, "Mining status." ); } diff --git a/mujina-miner/src/stratum_v2/client.rs b/mujina-miner/src/stratum_v2/client.rs new file mode 100644 index 00000000..4951370f --- /dev/null +++ b/mujina-miner/src/stratum_v2/client.rs @@ -0,0 +1,629 @@ +//! Encrypted Stratum V2 Extended Channel client. +//! +//! The client performs the full SV2 connection sequence: +//! DNS resolve → TCP connect → Noise NX handshake → SetupConnection → +//! OpenExtendedMiningChannel → main select! event loop. +//! +//! Read/write halves are decoupled via a spawned reader task because +//! `NoiseTcpReadHalf::read_frame()` is not cancellation-safe. +//! +//! # SV2 Spec References +//! +//! - [Common Protocol > `SetupConnection`][sv2-setup] +//! - [Mining Protocol > `OpenExtendedMiningChannel`][sv2-open-channel] +//! - [Protocol Security > URL Scheme and Pool Authority Key][sv2-url] +//! +//! [sv2-setup]: https://github.com/stratum-mining/sv2-spec/blob/main/03-Protocol-Overview.md#361-setupconnection-client---server +//! [sv2-open-channel]: https://github.com/stratum-mining/sv2-spec/blob/main/05-Mining-Protocol.md#534-openextendedminingchannel-client---server +//! [sv2-url]: https://github.com/stratum-mining/sv2-spec/blob/main/04-Protocol-Security.md#47-url-scheme-and-pool-authority-key + +use std::ops::ControlFlow; +use std::time::Duration; + +use bitcoin::pow::Target; +use stratum_apps::key_utils::Secp256k1PublicKey; +use stratum_apps::network_helpers::Error as NetworkError; +use stratum_apps::network_helpers::connect_with_noise; +use stratum_apps::network_helpers::noise_stream::{NoiseTcpReadHalf, NoiseTcpWriteHalf}; +use stratum_apps::network_helpers::resolve_host; +use stratum_apps::stratum_core::binary_sv2::Str0255; +use stratum_apps::stratum_core::codec_sv2::StandardEitherFrame; +use stratum_apps::stratum_core::common_messages_sv2::{ + ChannelEndpointChanged, Protocol, Reconnect, SetupConnection, +}; +use stratum_apps::stratum_core::mining_sv2::{ + CloseChannel, NewExtendedMiningJob, OpenExtendedMiningChannel, + OpenExtendedMiningChannelSuccess, SetNewPrevHash, SetTarget, SubmitSharesError, + SubmitSharesExtended, SubmitSharesSuccess, +}; +use stratum_apps::stratum_core::parsers_sv2::{AnyMessage, CommonMessages, Mining}; +use tokio::sync::mpsc; +use tokio_util::sync::CancellationToken; + +use crate::tracing::prelude::*; +use crate::types::HashRate; + +use super::error::{StratumV2Error, StratumV2Result}; + +/// Protocol constants for the Stratum V2 Extended Channel client. +pub mod constants { + /// Minimum total extranonce size (in bytes) requested from the pool. + /// + /// The pool splits this into a fixed prefix and a miner-controlled + /// set of bytes. Hardware rolls the 32-bit nonce at full speed; when + /// exhausted it wraps the rollable bytes for a fresh coinbase. A small + /// rollable bytes space wraps fast, causing share collisions at high + /// hashrates. If the pool assigns fewer bytes than this minimum, the + /// channel is rejected. + pub const MIN_EXTRANONCE_SIZE: usize = 8; + + /// Maximum total extranonce size (in bytes) accepted from the pool. + /// + /// `SubmitSharesExtended.extranonce` is typed `B032` (max 32 bytes), so + /// any pool-assigned `extranonce_size > 32` would cause every share to + /// fail `B032` encoding. Reject the channel at open time instead. + pub const MAX_EXTRANONCE_SIZE: usize = 32; +} + +/// Stratum V2 pool connection configuration. +#[derive(Debug, Clone)] +pub struct PoolConfig { + pub(crate) host: Str0255<'static>, + pub(crate) port: u16, + pub(crate) authority_pubkey: Secp256k1PublicKey, + pub(crate) user_identity: Str0255<'static>, + pub(crate) vendor: Str0255<'static>, + pub(crate) hardware_version: Str0255<'static>, + pub(crate) firmware: Str0255<'static>, + pub(crate) device_id: Str0255<'static>, + pub(crate) nominal_hash_rate: HashRate, +} + +impl PoolConfig { + #[expect( + clippy::too_many_arguments, + reason = "SV2 SetupConnection + OpenExtendedMiningChannel \ + field set; grouping into a sub-struct would not reduce the \ + caller's burden" + )] + pub fn new( + host: String, + port: u16, + authority_pubkey: Secp256k1PublicKey, + user_identity: String, + vendor: String, + hardware_version: String, + firmware: String, + device_id: String, + nominal_hash_rate: HashRate, + ) -> StratumV2Result { + Ok(Self { + host: Str0255::try_from(host).map_err(protocol_error)?, + port, + authority_pubkey, + user_identity: Str0255::try_from(user_identity).map_err(protocol_error)?, + vendor: Str0255::try_from(vendor).map_err(protocol_error)?, + hardware_version: Str0255::try_from(hardware_version).map_err(protocol_error)?, + firmware: Str0255::try_from(firmware).map_err(protocol_error)?, + device_id: Str0255::try_from(device_id).map_err(protocol_error)?, + nominal_hash_rate, + }) + } + + /// The pool host as text, for DNS resolution and logging. `Str0255`'s own + /// formatting renders the raw wire bytes, so callers that need the + /// hostname string go through here. + pub(crate) fn host(&self) -> String { + self.host.as_utf8_or_hex() + } +} + +/// Commands sent to the SV2 client from the consumer. +#[derive(Debug, Clone)] +pub enum ClientCommand { + SubmitShare(SubmitSharesExtended<'static>), +} + +/// Events emitted by the SV2 client to the consumer. +#[derive(Debug, Clone)] +pub enum ClientEvent { + /// Pool accepted SetupConnection. + SetupConnectionSuccess { used_version: u16, flags: u32 }, + /// Pool accepted OpenExtendedMiningChannel. + OpenExtendedMiningChannelSuccess(OpenExtendedMiningChannelSuccess<'static>), + /// New mining job from the pool. + NewExtendedMiningJob(NewExtendedMiningJob<'static>), + /// New previous block hash (invalidates un-activated future jobs). + SetNewPrevHash(SetNewPrevHash<'static>), + /// Pool updated the share difficulty target. + SetTarget(SetTarget<'static>), + /// Pool accepted a previously submitted share. + SubmitSharesSuccess(SubmitSharesSuccess), + /// Pool rejected a previously submitted share. + SubmitSharesError(SubmitSharesError<'static>), + /// Pool requested a reconnect. + Reconnect(Reconnect<'static>), + /// Pool reassigned the channel to a different endpoint. + ChannelEndpointChanged(ChannelEndpointChanged), + /// Pool closed the Extended Channel. + CloseChannel(CloseChannel<'static>), +} + +/// Outcome returned by [`StratumV2Client::run`] on a clean exit. +/// +/// Lets the caller distinguish a user-requested shutdown from a connection +/// close without relying on a side-channel flag or error-string inspection. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ClientOutcome { + /// Cancellation token was fired; miner requested a clean stop. + Shutdown, + /// TCP connection closed (pool side or network); caller should reconnect. + ConnectionClosed, + /// Pool sent `Reconnect`, `CloseChannel`, or `ChannelEndpointChanged`; + /// caller should reconnect (possibly to a new host). + PoolRequestedReconnect, +} + +/// Internal channel message from the reader task to the event loop. +/// +/// `NoiseTcpReadHalf::read_frame()` is not cancellation-safe: if dropped +/// mid-read, the internal codec state is left inconsistent. The fix is to +/// isolate it in a dedicated spawned task so it is never in a `select!` +/// branch that can be cancelled between bytes. This is the same pattern used +/// by `stratum-apps` itself in `Connection::spawn_reader` +/// (`network_helpers/noise_connection.rs`). +enum ReaderMessage { + Message(AnyMessage<'static>), + Error(StratumV2Error), + Done, +} + +/// Converts a `binary_sv2` specific error type to a our error type +/// so call sites can use `?` on the crate's own `Str0255::try_from`. +fn protocol_error(e: stratum_apps::stratum_core::binary_sv2::Error) -> StratumV2Error { + StratumV2Error::Protocol(format!("invalid SV2 value: {e:?}")) +} + +/// Encrypted Stratum V2 client using Extended Channels. +pub struct StratumV2Client { + config: PoolConfig, + event_tx: mpsc::Sender, + command_rx: mpsc::Receiver, + shutdown: CancellationToken, +} + +impl StratumV2Client { + pub fn new( + config: PoolConfig, + event_tx: mpsc::Sender, + command_rx: mpsc::Receiver, + shutdown: CancellationToken, + ) -> Self { + Self { + config, + event_tx, + command_rx, + shutdown, + } + } + + /// Connect to the pool and run the full protocol lifecycle. + pub async fn run(mut self) -> StratumV2Result { + const TCP_CONNECT_TIMEOUT_SECS: u64 = 5; + const NOISE_HANDSHAKE_TIMEOUT_SECS: u64 = 5; + + let addr = resolve_host(&self.config.host(), self.config.port) + .await + .map_err(|e| StratumV2Error::DnsResolutionFailed { + host: self.config.host(), + error: e.to_string(), + })?; + + debug!(host = %self.config.host(), %addr, "Resolved pool address"); + + // TCP connect + let stream = tokio::time::timeout( + Duration::from_secs(TCP_CONNECT_TIMEOUT_SECS), + tokio::net::TcpStream::connect(addr), + ) + .await + .map_err(|_| StratumV2Error::ConnectionFailed { + addr, + source: std::io::Error::new(std::io::ErrorKind::TimedOut, "TCP connect timeout"), + })? + .map_err(|source| StratumV2Error::ConnectionFailed { addr, source })?; + + debug!(host = %self.config.host(), "TCP connection established"); + + // Noise NX handshake (via stratum-apps) + let noise_stream = tokio::time::timeout( + Duration::from_secs(NOISE_HANDSHAKE_TIMEOUT_SECS), + connect_with_noise::>(stream, Some(self.config.authority_pubkey)), + ) + .await + .map_err(|_| StratumV2Error::Protocol("Noise handshake timed out".to_string()))? + .map_err(|e| StratumV2Error::Protocol(format!("Noise handshake failed: {e}")))?; + + info!(host = %self.config.host(), "Noise NX handshake completed"); + + // Split + spawn reader task + let (read_half, write_half) = noise_stream.into_split(); + + let (reader_tx, mut reader_rx) = mpsc::channel::(16); + let reader_shutdown = self.shutdown.clone(); + tokio::spawn(async move { + reader_task(read_half, reader_tx, reader_shutdown).await; + }); + + // Negotiate session, returning the write half for the event loop. + let write_half = match self.negotiate_session(&mut reader_rx, write_half).await { + Ok(wh) => wh, + Err(StratumV2Error::ReconnectDuringSetup) => { + return Ok(ClientOutcome::PoolRequestedReconnect); + } + Err(e) => return Err(e), + }; + + info!(host = %self.config.host(), user = %self.config.user_identity.as_utf8_or_hex(), "SV2 connection established"); + + // Main event loop + self.run_event_loop(reader_rx, write_half).await + } + + /// Negotiate SetupConnection + OpenExtendedMiningChannel. + /// + /// Returns the write half on success. If the pool sends a Reconnect + /// during setup, emits the event and returns `Err(ReconnectDuringSetup)` + /// so the caller can tear down and reconnect. + async fn negotiate_session( + &self, + reader_rx: &mut mpsc::Receiver, + mut write_half: NoiseTcpWriteHalf>, + ) -> StratumV2Result>> { + let setup = SetupConnection { + protocol: Protocol::MiningProtocol, + min_version: 2, + max_version: 2, + flags: 0, + endpoint_host: self.config.host.clone(), + endpoint_port: self.config.port, + vendor: self.config.vendor.clone(), + hardware_version: self.config.hardware_version.clone(), + firmware: self.config.firmware.clone(), + device_id: self.config.device_id.clone(), + }; + + debug!(host = %self.config.host(), "Sending SetupConnection"); + send_message( + &mut write_half, + AnyMessage::Common(CommonMessages::SetupConnection(setup.into_static())), + ) + .await?; + + match recv_one(reader_rx).await? { + AnyMessage::Common(CommonMessages::SetupConnectionSuccess(msg)) => { + debug!(host = %self.config.host(), version = msg.used_version, "SetupConnection accepted"); + if msg.used_version != 2 { + return Err(StratumV2Error::Protocol(format!( + "negotiated SV2 version {} (expected 2)", + msg.used_version + ))); + } + self.emit(ClientEvent::SetupConnectionSuccess { + used_version: msg.used_version, + flags: msg.flags, + }) + .await?; + } + AnyMessage::Common(CommonMessages::SetupConnectionError(msg)) => { + let reason = msg.error_code.as_utf8_or_hex(); + warn!(host = %self.config.host(), %reason, "SetupConnection rejected"); + return Err(StratumV2Error::SetupRejected(reason)); + } + unexpected => { + return Err(StratumV2Error::Protocol(format!( + "expected SetupConnectionSuccess, got {unexpected:?}" + ))); + } + } + + let max_target_bytes = Target::MAX.to_le_bytes(); + let max_target = stratum_apps::stratum_core::binary_sv2::U256::from(max_target_bytes); + + let open = OpenExtendedMiningChannel { + request_id: 1, + user_identity: self.config.user_identity.clone(), + nominal_hash_rate: self.config.nominal_hash_rate.0 as f32, + max_target, + min_extranonce_size: constants::MIN_EXTRANONCE_SIZE as u16, + }; + + debug!(host = %self.config.host(), "Sending OpenExtendedMiningChannel"); + send_message( + &mut write_half, + AnyMessage::Mining(Mining::OpenExtendedMiningChannel(open.into_static())), + ) + .await?; + + loop { + match recv_one(reader_rx).await? { + AnyMessage::Mining(Mining::OpenExtendedMiningChannelSuccess(msg)) => { + let target = target_from_le_bytes(msg.target.inner_as_ref())?; + debug!( + host = %self.config.host(), + channel_id = msg.channel_id, + extranonce_prefix = %hex::encode(msg.extranonce_prefix.inner_as_ref()), + %target, + "OpenExtendedMiningChannel accepted" + ); + + // extranonce_size covers the prefix and the rollable bytes areas + // of the Extended Extranonce; extranonce_prefix (already allocated by the + // upstream server) is separate. + if (msg.extranonce_size as usize) < constants::MIN_EXTRANONCE_SIZE + || (msg.extranonce_size as usize) > constants::MAX_EXTRANONCE_SIZE + { + return Err(StratumV2Error::ExtranonceSizeMismatch); + } + if msg.request_id != 1 { + return Err(StratumV2Error::Protocol(format!( + "unexpected request_id {}", + msg.request_id + ))); + } + + self.emit(ClientEvent::OpenExtendedMiningChannelSuccess(msg)) + .await?; + break; + } + AnyMessage::Mining(Mining::OpenMiningChannelError(msg)) => { + let reason = msg.error_code.as_utf8_or_hex(); + warn!(host = %self.config.host(), %reason, "OpenExtendedMiningChannel rejected"); + return Err(StratumV2Error::OpenChannelRejected(reason)); + } + AnyMessage::Common(CommonMessages::Reconnect(msg)) => { + info!(host = %self.config.host(), "Pool requested reconnect during setup"); + self.emit(ClientEvent::Reconnect(msg)).await?; + return Err(StratumV2Error::ReconnectDuringSetup); + } + // Pools may pipeline SetTarget, SetNewPrevHash, or NewExtendedMiningJob + // before the channel-open response arrives. Skip them here; the event + // loop processes them once the channel is established. + AnyMessage::Mining( + Mining::SetTarget(_) + | Mining::SetNewPrevHash(_) + | Mining::NewExtendedMiningJob(_), + ) => { + debug!(host = %self.config.host(), "Skipping pipelined mining message during channel open"); + } + unexpected => { + return Err(StratumV2Error::Protocol(format!( + "expected OpenExtendedMiningChannelSuccess, got {unexpected:?}" + ))); + } + } + } + + Ok(write_half) + } + + /// Main event loop: dispatch commands and route inbound messages. + async fn run_event_loop( + &mut self, + mut reader_rx: mpsc::Receiver, + mut write_half: NoiseTcpWriteHalf>, + ) -> StratumV2Result { + loop { + tokio::select! { + reader_msg = reader_rx.recv() => { + match reader_msg { + Some(ReaderMessage::Message(msg)) => { + if self.handle_message(msg).await?.is_break() { + info!(host = %self.config.host(), "Pool requested disconnect"); + return Ok(ClientOutcome::PoolRequestedReconnect); + } + } + Some(ReaderMessage::Error(e)) => { + error!(host = %self.config.host(), err = %e, "Reader task error"); + return Err(e); + } + Some(ReaderMessage::Done) | None => { + info!(host = %self.config.host(), "Reader task finished"); + return Ok(ClientOutcome::ConnectionClosed); + } + } + } + + command = self.command_rx.recv() => { + match command { + Some(ClientCommand::SubmitShare(share)) => { + trace!(job_id = share.job_id, "Submitting share"); + send_message( + &mut write_half, + AnyMessage::Mining(Mining::SubmitSharesExtended(share)), + ).await?; + } + None => { + info!(host = %self.config.host(), "Command channel closed; stopping"); + return Ok(ClientOutcome::Shutdown); + } + } + } + + _ = self.shutdown.cancelled() => { + info!(host = %self.config.host(), "Shutting down"); + return Ok(ClientOutcome::Shutdown); + } + } + } + } + + /// Dispatch a received message. + /// + /// Returns `Ok(ControlFlow::Break(()))` when the pool requests a disconnect + /// (Reconnect, ChannelEndpointChanged, CloseChannel), `Ok(ControlFlow::Continue(()))` + /// otherwise. + async fn handle_message( + &mut self, + msg: AnyMessage<'static>, + ) -> StratumV2Result> { + match msg { + AnyMessage::Mining(Mining::NewExtendedMiningJob(job)) => { + trace!( + job_id = job.job_id, + future = job.is_future(), + "NewExtendedMiningJob" + ); + self.emit(ClientEvent::NewExtendedMiningJob(job)).await?; + Ok(ControlFlow::Continue(())) + } + AnyMessage::Mining(Mining::SetNewPrevHash(prev)) => { + debug!(prev_hash = %hex::encode(prev.prev_hash.inner_as_ref()), "SetNewPrevHash"); + self.emit(ClientEvent::SetNewPrevHash(prev)).await?; + Ok(ControlFlow::Continue(())) + } + AnyMessage::Mining(Mining::SetTarget(target)) => { + debug!(channel_id = target.channel_id, "SetTarget"); + self.emit(ClientEvent::SetTarget(target)).await?; + Ok(ControlFlow::Continue(())) + } + AnyMessage::Mining(Mining::SubmitSharesSuccess(success)) => { + trace!(channel_id = success.channel_id, "SubmitShares.Success"); + self.emit(ClientEvent::SubmitSharesSuccess(success)).await?; + Ok(ControlFlow::Continue(())) + } + AnyMessage::Mining(Mining::SubmitSharesError(error)) => { + warn!( + channel_id = error.channel_id, + seq = error.sequence_number, + reason = error.error_code.as_utf8_or_hex(), + "SubmitShares.Error" + ); + self.emit(ClientEvent::SubmitSharesError(error)).await?; + Ok(ControlFlow::Continue(())) + } + AnyMessage::Common(CommonMessages::Reconnect(msg)) => { + info!(host = %self.config.host(), "Reconnect"); + self.emit(ClientEvent::Reconnect(msg)).await?; + Ok(ControlFlow::Break(())) + } + AnyMessage::Common(CommonMessages::ChannelEndpointChanged(msg)) => { + info!(host = %self.config.host(), "ChannelEndpointChanged"); + self.emit(ClientEvent::ChannelEndpointChanged(msg)).await?; + Ok(ControlFlow::Break(())) + } + AnyMessage::Mining(Mining::CloseChannel(msg)) => { + info!(channel_id = msg.channel_id, "CloseChannel"); + self.emit(ClientEvent::CloseChannel(msg)).await?; + Ok(ControlFlow::Break(())) + } + unexpected => { + warn!(host = %self.config.host(), ?unexpected, "Ignoring unexpected message"); + Ok(ControlFlow::Continue(())) + } + } + } + + async fn emit(&self, event: ClientEvent) -> StratumV2Result<()> { + self.event_tx + .send(event) + .await + .map_err(|_| StratumV2Error::Protocol("event receiver dropped".to_string())) + } +} + +async fn reader_task( + mut read_half: NoiseTcpReadHalf>, + tx: mpsc::Sender, + shutdown: CancellationToken, +) { + loop { + tokio::select! { + result = read_half.read_frame() => { + match result { + Ok(frame) => { + let msg = match extract_any_message(frame) { + Ok(m) => m, + Err(e) => { + let _ = tx.send(ReaderMessage::Error(e)).await; + return; + } + }; + if tx.send(ReaderMessage::Message(msg)).await.is_err() { + return; + } + } + Err(e) => { + if matches!(e, NetworkError::SocketClosed) { + debug!("Pool closed connection"); + let _ = tx.send(ReaderMessage::Done).await; + } else { + warn!(err = %e, "Read error from pool"); + let _ = tx.send(ReaderMessage::Error( + StratumV2Error::Protocol(format!("read error: {e}")) + )).await; + } + return; + } + } + } + _ = shutdown.cancelled() => { + let _ = tx.send(ReaderMessage::Done).await; + return; + } + } + } +} + +fn extract_any_message( + frame: StandardEitherFrame>, +) -> StratumV2Result> { + match frame { + StandardEitherFrame::Sv2(mut sv2_frame) => { + let header = sv2_frame + .get_header() + .ok_or_else(|| StratumV2Error::Protocol("frame without header".to_string()))?; + AnyMessage::try_from((header, sv2_frame.payload())) + .map(|m| m.into_static()) + .map_err(|e| StratumV2Error::Protocol(format!("parse error: {e}"))) + } + StandardEitherFrame::HandShake(_) => Err(StratumV2Error::Protocol( + "unexpected handshake frame after Noise handshake".to_string(), + )), + } +} + +async fn send_message( + write_half: &mut NoiseTcpWriteHalf>, + msg: AnyMessage<'static>, +) -> StratumV2Result<()> { + let sv2_frame = + msg.try_into() + .map_err(|e: stratum_apps::stratum_core::parsers_sv2::ParserError| { + StratumV2Error::Protocol(format!("frame encode failed: {e}")) + })?; + write_half + .write_frame(StandardEitherFrame::Sv2(sv2_frame)) + .await + .map_err(|e| StratumV2Error::Protocol(format!("write_frame failed: {e}"))) +} + +async fn recv_one(rx: &mut mpsc::Receiver) -> StratumV2Result> { + match rx.recv().await { + Some(ReaderMessage::Message(msg)) => Ok(msg), + Some(ReaderMessage::Error(e)) => Err(e), + Some(ReaderMessage::Done) | None => Err(StratumV2Error::Protocol( + "reader terminated during negotiation".to_string(), + )), + } +} + +pub(crate) fn target_from_le_bytes(bytes: &[u8]) -> StratumV2Result { + let arr: [u8; 32] = bytes.try_into().map_err(|_| { + StratumV2Error::Protocol(format!( + "expected 32-byte target, got {} bytes", + bytes.len() + )) + })?; + Ok(Target::from_le_bytes(arr)) +} diff --git a/mujina-miner/src/stratum_v2/error.rs b/mujina-miner/src/stratum_v2/error.rs new file mode 100644 index 00000000..e8c13a76 --- /dev/null +++ b/mujina-miner/src/stratum_v2/error.rs @@ -0,0 +1,91 @@ +//! Error types for Stratum V2 protocol. +//! +//! Variants are classified as fatal (misconfigured key, explicit pool rejection) +//! or transient (I/O, timeout) so callers can decide whether to retry or abort. + +use std::net::SocketAddr; + +use thiserror::Error; + +/// Errors that can occur during Stratum V2 client operation. +#[derive(Error, Debug)] +pub enum StratumV2Error { + /// DNS resolution or TCP connection failure. + /// + /// Transient — may resolve on retry (DNS propagation, pool restart). + #[error("connection to {addr} failed: {source}")] + ConnectionFailed { + addr: SocketAddr, + #[source] + source: std::io::Error, + }, + + /// Pool rejected our `SetupConnection` message. + /// + /// Fatal — indicates protocol version or capability mismatch that won't + /// resolve without a configuration change. + #[error("pool rejected setup connection: {0}")] + SetupRejected(String), + + /// Pool rejected our `OpenExtendedMiningChannel` message. + /// + /// Fatal — typically caused by invalid user identity or unsupported + /// extranonce size request. + #[error("pool rejected open channel: {0}")] + OpenChannelRejected(String), + + /// Pool-assigned extranonce size is outside `[MIN_EXTRANONCE_SIZE, MAX_EXTRANONCE_SIZE]`. + #[error("pool assigned extranonce size outside acceptable range")] + ExtranonceSizeMismatch, + + /// Authority public key is invalid for Noise handshake. + /// + /// Fatal — the configured key is malformed. No retry can fix this. + #[error("invalid authority public key: {0}")] + InvalidAuthorityKey(String), + + /// DNS resolution failed. + /// + /// Transient — may resolve on retry. + #[error("DNS resolution failed for {host}: {error}")] + DnsResolutionFailed { host: String, error: String }, + + /// Network or framed I/O error. + /// + /// Transient — triggers reconnect. + #[error("I/O error: {0}")] + Io(#[from] std::io::Error), + + /// Protocol-level error (unexpected message, serialization failure). + #[error("protocol error: {0}")] + Protocol(String), + + /// Client was shut down via cancellation token. + #[error("client shutdown")] + Shutdown, + + /// Pool sent `Reconnect` before channel setup completed; caller should reconnect. + #[error("pool requested reconnect during setup")] + ReconnectDuringSetup, +} + +impl StratumV2Error { + /// Whether this error is unrecoverable and should not be retried. + /// + /// Authorization failures, key errors, and explicit pool rejections are + /// fatal — they won't fix themselves without a configuration change. + /// Everything else (network errors, timeouts, DNS failures, extranonce + /// size mismatches) may resolve on retry or reconnect. + pub fn is_fatal(&self) -> bool { + matches!( + self, + StratumV2Error::SetupRejected(_) + | StratumV2Error::OpenChannelRejected(_) + | StratumV2Error::InvalidAuthorityKey(_) + | StratumV2Error::ExtranonceSizeMismatch + ) + } +} + +/// Convenient Result type for Stratum V2 operations. +pub type StratumV2Result = Result; diff --git a/mujina-miner/src/stratum_v2/integration_tests.rs b/mujina-miner/src/stratum_v2/integration_tests.rs new file mode 100644 index 00000000..c882cf0a --- /dev/null +++ b/mujina-miner/src/stratum_v2/integration_tests.rs @@ -0,0 +1,726 @@ +//! Integration tests for [`StratumV2Source`]. +//! +//! Each `#[ignore]` test spins up a real SV2 pool behind a [`Sniffer`] proxy +//! and asserts on the message exchange at the sniffer boundary. +//! `integration_tests_sv2` downloads Bitcoin Core and the SV2 template +//! provider binary on first run (~200 MB, cached in `~/.cargo/`). +//! +//! Use `--test-threads=4` to avoid port-binding races from concurrent `bitcoind` +//! instances when running the full suite: +//! +//! ```text +//! cargo test -p mujina-miner -- stratum_v2::integration_tests --ignored --nocapture --test-threads=4 +//! ``` + +use integration_tests_sv2::{ + interceptor::{IgnoreMessage, MessageDirection, ReplaceMessage}, + template_provider::DifficultyLevel, + *, +}; +use stratum_apps::{ + key_utils::Secp256k1PublicKey, + stratum_core::{ + common_messages_sv2::*, + mining_sv2::*, + parsers_sv2::{AnyMessage, CommonMessages, Mining}, + }, +}; +use tokio::sync::mpsc; +use tokio_util::sync::CancellationToken; + +use crate::{ + job_source::{ + Extranonce2, MerkleRootKind, Share, SourceCommand, SourceEvent, stratum_v2::StratumV2Source, + }, + stratum_v2::PoolConfig, + types::HashRate, +}; + +// --------------------------------------------------------------------------- +// Helpers +// --------------------------------------------------------------------------- + +/// Well-known authority public key used by the Sniffer's Noise responder. +/// +/// Our client must be configured with this key so the Noise handshake +/// succeeds when connecting to the sniffer address instead of the real pool. +const SNIFFER_AUTHORITY_PUBKEY: &str = "9auqWEzQDVyd2oe1JVGFLMLHZtCo2FFqZwtKA5gd9xbuEu7PH72"; + +/// Build a [`PoolConfig`] pointing at `addr` with the Sniffer authority key. +fn pool_config(addr: std::net::SocketAddr) -> PoolConfig { + let authority_pubkey = SNIFFER_AUTHORITY_PUBKEY + .parse::() + .expect("test authority pubkey must parse"); + PoolConfig::new( + addr.ip().to_string(), + addr.port(), + authority_pubkey, + "test-worker".to_string(), + "test-vendor".to_string(), + "1.0".to_string(), + "1.0".to_string(), + String::new(), + HashRate::from_megahashes(1.0), + ) + .expect("PoolConfig fields are within Str0255 limits") +} + +/// Spawn a [`StratumV2Source`] and pre-seed it with a non-zero hash-rate so +/// it proceeds to connect without waiting for an `UpdateHashRate` command. +/// +/// Returns `(cmd_tx, event_rx, join_handle)`. +fn spawn_source( + addr: std::net::SocketAddr, + shutdown: CancellationToken, +) -> ( + mpsc::Sender, + mpsc::Receiver, + tokio::task::JoinHandle>, +) { + let (cmd_tx, cmd_rx) = mpsc::channel(10); + let (event_tx, event_rx) = mpsc::channel(100); + + // Pre-buffer a non-zero hashrate so the source's startup loop breaks + // immediately when it first polls the command channel. + cmd_tx + .try_send(SourceCommand::UpdateHashRate(HashRate::from_megahashes( + 1.0, + ))) + .expect("channel must have capacity for one message"); + + let source = StratumV2Source::new(pool_config(addr), cmd_rx, event_tx, shutdown); + let handle = tokio::spawn(async move { source.run().await }); + (cmd_tx, event_rx, handle) +} + +// --------------------------------------------------------------------------- +// Tests +// --------------------------------------------------------------------------- + +/// A cancelled shutdown token causes [`StratumV2Source::run`] to return +/// `Ok(())` cleanly before the source ever connects to a pool. +/// +/// This test requires no external binaries. +#[tokio::test] +async fn clean_shutdown_before_connect() { + let shutdown = CancellationToken::new(); + let (_cmd_tx, cmd_rx) = mpsc::channel(10); + let (event_tx, _event_rx) = mpsc::channel(100); + + let authority_pubkey = SNIFFER_AUTHORITY_PUBKEY + .parse::() + .expect("test key must parse"); + let config = PoolConfig::new( + "127.0.0.1".to_string(), + 1, // never connected to + authority_pubkey, + "w".to_string(), + "v".to_string(), + "h".to_string(), + "f".to_string(), + String::new(), + HashRate::default(), // zero → source stays in startup loop + ) + .unwrap(); + + let source = StratumV2Source::new(config, cmd_rx, event_tx, shutdown.clone()); + let handle = tokio::spawn(async move { source.run().await }); + + // Cancel while the source is still waiting for a non-zero hashrate. + shutdown.cancel(); + + let result = handle.await.expect("task must not panic"); + assert!(result.is_ok(), "expected Ok(()), got {result:?}"); +} + +/// [`StratumV2Source`] sends `SetupConnection` with `protocol = +/// MiningProtocol`, `min_version = 2`, `max_version = 2`, and `flags = 0` +/// (no `requires_standard_job` flag) as its first message. +#[tokio::test] +#[ignore = "downloads Bitcoin Core and sv2-tp binaries on first run"] +async fn setup_connection_uses_mining_protocol() { + start_tracing(); + + let (_tp, tp_addr) = start_template_provider(None, DifficultyLevel::Low); + let (_pool, pool_addr, _) = start_pool(sv2_tp_config(tp_addr), vec![], vec![], false).await; + let (sniffer, sniffer_addr) = + start_sniffer("setup-connection", pool_addr, false, vec![], Some(30)); + + let shutdown = CancellationToken::new(); + let (_cmd_tx, _event_rx, _handle) = spawn_source(sniffer_addr, shutdown.clone()); + + sniffer + .wait_for_message_type(MessageDirection::ToUpstream, MESSAGE_TYPE_SETUP_CONNECTION) + .await; + + assert_common_message!( + &sniffer.next_message_from_downstream(), + SetupConnection, + protocol, + Protocol::MiningProtocol, + min_version, + 2u16, + max_version, + 2u16, + flags, + 0u32 + ); + + shutdown.cancel(); +} + +/// After `SetupConnectionSuccess`, the source opens an Extended Mining Channel +/// and the pool replies with `OpenExtendedMiningChannelSuccess`. +#[tokio::test] +#[ignore = "downloads Bitcoin Core and sv2-tp binaries on first run"] +async fn extended_channel_opened_after_setup() { + start_tracing(); + + let (_tp, tp_addr) = start_template_provider(None, DifficultyLevel::Low); + let (_pool, pool_addr, _) = start_pool(sv2_tp_config(tp_addr), vec![], vec![], false).await; + let (sniffer, sniffer_addr) = start_sniffer("channel-open", pool_addr, false, vec![], Some(30)); + + let shutdown = CancellationToken::new(); + let (_cmd_tx, _event_rx, _handle) = spawn_source(sniffer_addr, shutdown.clone()); + + // Drain SetupConnection so the next pop returns OpenExtendedMiningChannel. + sniffer + .wait_for_message_type_and_clean_queue( + MessageDirection::ToUpstream, + MESSAGE_TYPE_SETUP_CONNECTION, + ) + .await; + + sniffer + .wait_for_message_type( + MessageDirection::ToUpstream, + MESSAGE_TYPE_OPEN_EXTENDED_MINING_CHANNEL, + ) + .await; + assert_mining_message!( + &sniffer.next_message_from_downstream(), + OpenExtendedMiningChannel + ); + + // Drain SetupConnectionSuccess so the next pop returns OpenExtendedMiningChannelSuccess. + sniffer + .wait_for_message_type_and_clean_queue( + MessageDirection::ToDownstream, + MESSAGE_TYPE_SETUP_CONNECTION_SUCCESS, + ) + .await; + + sniffer + .wait_for_message_type( + MessageDirection::ToDownstream, + MESSAGE_TYPE_OPEN_EXTENDED_MINING_CHANNEL_SUCCESS, + ) + .await; + assert_mining_message!( + &sniffer.next_message_from_upstream(), + OpenExtendedMiningChannelSuccess + ); + + shutdown.cancel(); +} + +/// [`StratumV2Source`] emits [`SourceEvent::ReplaceJob`] after the pool delivers +/// a `NewExtendedMiningJob` / `SetNewPrevHash` pair. +#[tokio::test] +#[ignore = "downloads Bitcoin Core and sv2-tp binaries on first run"] +async fn replace_job_emitted_on_job_and_prevhash() { + start_tracing(); + + let (_tp, tp_addr) = start_template_provider(None, DifficultyLevel::Low); + let (_pool, pool_addr, _) = start_pool(sv2_tp_config(tp_addr), vec![], vec![], false).await; + let (sniffer, sniffer_addr) = start_sniffer("job-flow", pool_addr, false, vec![], Some(60)); + + let shutdown = CancellationToken::new(); + let (_cmd_tx, mut event_rx, _handle) = spawn_source(sniffer_addr, shutdown.clone()); + + // Wait until the pool has delivered both halves of the job pair. + sniffer + .wait_for_message_type( + MessageDirection::ToDownstream, + MESSAGE_TYPE_NEW_EXTENDED_MINING_JOB, + ) + .await; + sniffer + .wait_for_message_type( + MessageDirection::ToDownstream, + MESSAGE_TYPE_MINING_SET_NEW_PREV_HASH, + ) + .await; + + let got_replace_job = tokio::time::timeout(std::time::Duration::from_secs(10), async { + loop { + match event_rx.recv().await { + Some(SourceEvent::ReplaceJob(_)) => return true, + Some(_) => {} // skip ClearJobs, SharesAccepted, … + None => return false, // sender dropped + } + } + }) + .await + .unwrap_or(false); + + assert!( + got_replace_job, + "expected SourceEvent::ReplaceJob within 10 s" + ); + + shutdown.cancel(); +} + +/// `OpenExtendedMiningChannel` carries the worker name from [`PoolConfig`] as +/// `user_identity`, a positive nominal hash rate, and a non-zero +/// `min_extranonce_size`. +#[tokio::test] +#[ignore = "downloads Bitcoin Core and sv2-tp binaries on first run"] +async fn open_extended_mining_channel_fields() { + start_tracing(); + + let (_tp, tp_addr) = start_template_provider(None, DifficultyLevel::Low); + let (_pool, pool_addr, _) = start_pool(sv2_tp_config(tp_addr), vec![], vec![], false).await; + let (sniffer, sniffer_addr) = + start_sniffer("open-channel-fields", pool_addr, false, vec![], Some(30)); + + let shutdown = CancellationToken::new(); + let (_cmd_tx, _event_rx, _handle) = spawn_source(sniffer_addr, shutdown.clone()); + + // Drain SetupConnection first so the next pop is OpenExtendedMiningChannel. + sniffer + .wait_for_message_type_and_clean_queue( + MessageDirection::ToUpstream, + MESSAGE_TYPE_SETUP_CONNECTION, + ) + .await; + + sniffer + .wait_for_message_type( + MessageDirection::ToUpstream, + MESSAGE_TYPE_OPEN_EXTENDED_MINING_CHANNEL, + ) + .await; + + let msg = sniffer.next_message_from_downstream(); + match msg { + Some((_, AnyMessage::Mining(Mining::OpenExtendedMiningChannel(m)))) => { + assert_eq!( + m.user_identity.as_utf8_or_hex(), + "test-worker", + "user_identity must match pool config worker name" + ); + assert!( + m.nominal_hash_rate > 0.0, + "nominal_hash_rate must be positive" + ); + assert!( + m.min_extranonce_size >= 4, + "min_extranonce_size must be at least 4 bytes" + ); + } + other => panic!("expected OpenExtendedMiningChannel, got {:?}", other), + } + + shutdown.cancel(); +} + +/// A share for a job ID that was never assigned by the pool is silently dropped +/// by [`StratumV2Source`] — no `SubmitSharesExtended` must reach the pool. +#[tokio::test] +#[ignore = "downloads Bitcoin Core and sv2-tp binaries on first run"] +async fn stale_share_not_forwarded() { + start_tracing(); + + let (_tp, tp_addr) = start_template_provider(None, DifficultyLevel::Low); + let (_pool, pool_addr, _) = start_pool(sv2_tp_config(tp_addr), vec![], vec![], false).await; + let (sniffer, sniffer_addr) = start_sniffer("stale-share", pool_addr, false, vec![], Some(30)); + + let shutdown = CancellationToken::new(); + let (cmd_tx, _event_rx, _handle) = spawn_source(sniffer_addr, shutdown.clone()); + + // Wait for the channel to be fully open before sending any share. + sniffer + .wait_for_message_type( + MessageDirection::ToDownstream, + MESSAGE_TYPE_OPEN_EXTENDED_MINING_CHANNEL_SUCCESS, + ) + .await; + + // Send a share with a job ID unknown to the source; it will be dropped. + cmd_tx + .send(SourceCommand::SubmitShare(Share { + job_id: "99999".to_string(), + nonce: 0, + time: 0, + version: bitcoin::block::Version::from_consensus(0x2000_0000), + extranonce2: Some(Extranonce2::new(0, 4).expect("valid extranonce2")), + })) + .await + .expect("command channel must be open"); + + let not_forwarded = sniffer + .assert_message_not_present( + MessageDirection::ToUpstream, + MESSAGE_TYPE_SUBMIT_SHARES_EXTENDED, + std::time::Duration::from_secs(2), + ) + .await; + assert!( + not_forwarded, + "stale share must not be forwarded to the pool" + ); + + shutdown.cancel(); +} + +/// Cancelling the shutdown token while an Extended Mining Channel is active +/// causes [`StratumV2Source::run`] to return `Ok(())` cleanly. +#[tokio::test] +#[ignore = "downloads Bitcoin Core and sv2-tp binaries on first run"] +async fn shutdown_during_active_session() { + start_tracing(); + + let (_tp, tp_addr) = start_template_provider(None, DifficultyLevel::Low); + let (_pool, pool_addr, _) = start_pool(sv2_tp_config(tp_addr), vec![], vec![], false).await; + let (sniffer, sniffer_addr) = + start_sniffer("shutdown-active", pool_addr, false, vec![], Some(30)); + + let shutdown = CancellationToken::new(); + let (_cmd_tx, _event_rx, handle) = spawn_source(sniffer_addr, shutdown.clone()); + + // Wait until the channel is fully open before triggering shutdown. + sniffer + .wait_for_message_type( + MessageDirection::ToDownstream, + MESSAGE_TYPE_OPEN_EXTENDED_MINING_CHANNEL_SUCCESS, + ) + .await; + + shutdown.cancel(); + + let result = handle.await.expect("task must not panic"); + assert!(result.is_ok(), "expected Ok(()), got {result:?}"); +} + +/// Source receives a job, submits shares until the pool accepts one, and emits +/// [`SourceEvent::SharesAccepted`]. With [`DifficultyLevel::Low`] acceptance +/// almost always happens on the first nonce. +#[tokio::test] +#[ignore = "downloads Bitcoin Core and sv2-tp binaries on first run"] +async fn share_submission_accepted() { + start_tracing(); + + let (_tp, tp_addr) = start_template_provider(None, DifficultyLevel::Low); + let (_pool, pool_addr, _) = start_pool(sv2_tp_config(tp_addr), vec![], vec![], false).await; + let (sniffer, sniffer_addr) = + start_sniffer("share-accepted", pool_addr, false, vec![], Some(30)); + + let shutdown = CancellationToken::new(); + let (cmd_tx, mut event_rx, _handle) = spawn_source(sniffer_addr, shutdown.clone()); + + // Wait for ReplaceJob — proof the channel is open and a valid job is active. + let template = tokio::time::timeout(std::time::Duration::from_secs(30), async { + loop { + match event_rx.recv().await { + Some(SourceEvent::ReplaceJob(t)) => return t, + Some(_) => {} + None => panic!("event channel closed before ReplaceJob arrived"), + } + } + }) + .await + .expect("ReplaceJob must arrive within 30 s"); + + let en2_size = match &template.merkle_root { + MerkleRootKind::Computed(mrt) => mrt.extranonce2_range.size, + MerkleRootKind::Fixed(_) => panic!("SV2 source must produce a Computed merkle root"), + }; + + // Retry up to 32 nonces; DifficultyLevel::Low means ~255/256 succeed immediately. + let mut accepted = false; + for nonce in 0_u32..32 { + cmd_tx + .send(SourceCommand::SubmitShare(Share { + job_id: template.id.clone(), + nonce, + time: template.time, + version: template.version.base(), + extranonce2: Some(Extranonce2::new(0, en2_size).expect("valid extranonce2")), + })) + .await + .expect("command channel must be open"); + + let outcome = tokio::time::timeout(std::time::Duration::from_secs(5), async { + loop { + match event_rx.recv().await { + Some(SourceEvent::SharesAccepted(_)) => return Some(true), + Some(SourceEvent::SharesRejected) => return Some(false), + Some(_) => {} + None => return None, + } + } + }) + .await; + + match outcome { + Ok(Some(true)) => { + accepted = true; + break; + } + Ok(Some(false)) => {} // pool sent difficulty-too-low; try next nonce + Ok(None) => panic!("event channel closed before share response"), + Err(_) => {} // no response within 5 s; try next nonce + } + } + + assert!( + accepted, + "no share accepted within 32 nonces with DifficultyLevel::Low" + ); + + sniffer + .wait_for_message_type( + MessageDirection::ToUpstream, + MESSAGE_TYPE_SUBMIT_SHARES_EXTENDED, + ) + .await; + + shutdown.cancel(); +} + +/// When the sniffer drops all `NewExtendedMiningJob` messages before they reach +/// the source, `SetNewPrevHash` arrives with no matching future job buffered. +/// The source must emit `SourceEvent::ClearJobs` rather than crashing or +/// stalling. +#[tokio::test] +#[ignore = "downloads Bitcoin Core and sv2-tp binaries on first run"] +async fn set_new_prev_hash_with_no_future_job_emits_clear_jobs() { + start_tracing(); + + let (_tp, tp_addr) = start_template_provider(None, DifficultyLevel::Low); + let (_pool, pool_addr, _) = start_pool(sv2_tp_config(tp_addr), vec![], vec![], false).await; + + // Drop all NewExtendedMiningJob messages; SetNewPrevHash still flows through. + let drop_jobs = IgnoreMessage::new( + MessageDirection::ToDownstream, + MESSAGE_TYPE_NEW_EXTENDED_MINING_JOB, + ); + let (sniffer, sniffer_addr) = start_sniffer( + "no-future-job", + pool_addr, + false, + vec![drop_jobs.into()], + Some(60), + ); + + let shutdown = CancellationToken::new(); + let (_cmd_tx, mut event_rx, _handle) = spawn_source(sniffer_addr, shutdown.clone()); + + sniffer + .wait_for_message_type( + MessageDirection::ToDownstream, + MESSAGE_TYPE_MINING_SET_NEW_PREV_HASH, + ) + .await; + + let got_clear = tokio::time::timeout(std::time::Duration::from_secs(10), async { + loop { + match event_rx.recv().await { + Some(SourceEvent::ClearJobs) => return true, + Some(_) => {} + None => return false, + } + } + }) + .await + .unwrap_or(false); + + assert!( + got_clear, + "expected SourceEvent::ClearJobs when SetNewPrevHash arrives with no buffered job" + ); + + shutdown.cancel(); +} + +/// The sniffer replaces `SubmitSharesSuccess` with a crafted `SubmitSharesError`; +/// the source must emit [`SourceEvent::SharesRejected`]. +#[tokio::test] +#[ignore = "downloads Bitcoin Core and sv2-tp binaries on first run"] +async fn share_submission_rejected() { + start_tracing(); + + let (_tp, tp_addr) = start_template_provider(None, DifficultyLevel::Low); + let (_pool, pool_addr, _) = start_pool(sv2_tp_config(tp_addr), vec![], vec![], false).await; + + // channel_id = 2: sv2-apps assigns group_id = 1, first individual channel_id = 2. + let rejection = SubmitSharesError { + channel_id: 2, + sequence_number: 0, + error_code: "bad-share".to_string().try_into().unwrap(), + }; + let replace_success = ReplaceMessage::new( + MessageDirection::ToDownstream, + MESSAGE_TYPE_SUBMIT_SHARES_SUCCESS, + AnyMessage::Mining(Mining::SubmitSharesError(rejection)), + ); + let (sniffer, sniffer_addr) = start_sniffer( + "share-rejected", + pool_addr, + false, + vec![replace_success.into()], + Some(30), + ); + + let shutdown = CancellationToken::new(); + let (cmd_tx, mut event_rx, _handle) = spawn_source(sniffer_addr, shutdown.clone()); + + // Wait for a valid job before submitting a share. + let template = tokio::time::timeout(std::time::Duration::from_secs(60), async { + loop { + match event_rx.recv().await { + Some(SourceEvent::ReplaceJob(t)) => return t, + Some(_) => {} + None => panic!("event channel closed before ReplaceJob arrived"), + } + } + }) + .await + .expect("ReplaceJob must arrive within 60 s"); + + let en2_size = match &template.merkle_root { + MerkleRootKind::Computed(mrt) => mrt.extranonce2_range.size, + MerkleRootKind::Fixed(_) => panic!("SV2 source must produce a Computed merkle root"), + }; + + cmd_tx + .send(SourceCommand::SubmitShare(Share { + job_id: template.id.clone(), + nonce: 0, + time: template.time, + version: template.version.base(), + extranonce2: Some(Extranonce2::new(0, en2_size).expect("valid extranonce2")), + })) + .await + .expect("command channel must be open"); + + sniffer + .wait_for_message_type( + MessageDirection::ToUpstream, + MESSAGE_TYPE_SUBMIT_SHARES_EXTENDED, + ) + .await; + + let rejected = tokio::time::timeout(std::time::Duration::from_secs(15), async { + loop { + match event_rx.recv().await { + Some(SourceEvent::SharesRejected) => return true, + Some(SourceEvent::SharesAccepted(_)) => return false, // wrong path + Some(_) => {} + None => return false, + } + } + }) + .await + .unwrap_or(false); + + assert!(rejected, "expected SourceEvent::SharesRejected"); + + shutdown.cancel(); +} + +/// The sniffer advertises `extranonce_size = 16`; the source caps its internal +/// counter at 8 bytes and zero-pads to fill the allocation on the wire. +#[tokio::test] +#[ignore = "downloads Bitcoin Core and sv2-tp binaries on first run"] +async fn extranonce2_wire_length_matches_alloc() { + start_tracing(); + + let (_tp, tp_addr) = start_template_provider(None, DifficultyLevel::Low); + let (_pool, pool_addr, _) = start_pool(sv2_tp_config(tp_addr), vec![], vec![], false).await; + + // Replace OpenExtendedMiningChannelSuccess with one advertising extranonce_size=16 + // (miner's portion only; 4-byte prefix is separate). + let fake_success = OpenExtendedMiningChannelSuccess { + request_id: 1, // first channel open + channel_id: 2, // sv2-apps: group_id=1, first individual channel_id=2 + target: [0xff_u8; 32].to_vec().try_into().unwrap(), + extranonce_size: 16u16, + extranonce_prefix: vec![0u8; 4].try_into().unwrap(), + group_channel_id: 1, + }; + let replace_open = ReplaceMessage::new( + MessageDirection::ToDownstream, + MESSAGE_TYPE_OPEN_EXTENDED_MINING_CHANNEL_SUCCESS, + AnyMessage::Mining(Mining::OpenExtendedMiningChannelSuccess(fake_success)), + ); + let (sniffer, sniffer_addr) = start_sniffer( + "extranonce-padding", + pool_addr, + false, + vec![replace_open.into()], + Some(30), + ); + + let shutdown = CancellationToken::new(); + let (cmd_tx, mut event_rx, _handle) = spawn_source(sniffer_addr, shutdown.clone()); + + let template = tokio::time::timeout(std::time::Duration::from_secs(60), async { + loop { + match event_rx.recv().await { + Some(SourceEvent::ReplaceJob(t)) => return t, + Some(_) => {} + None => panic!("event channel closed before ReplaceJob arrived"), + } + } + }) + .await + .expect("ReplaceJob must arrive within 60 s"); + + let en2_size = match &template.merkle_root { + MerkleRootKind::Computed(mrt) => mrt.extranonce2_range.size, + MerkleRootKind::Fixed(_) => panic!("SV2 source must produce a Computed merkle root"), + }; + + cmd_tx + .send(SourceCommand::SubmitShare(Share { + job_id: template.id.clone(), + nonce: 0, + time: template.time, + version: template.version.base(), + extranonce2: Some(Extranonce2::new(0, en2_size).expect("valid extranonce2")), + })) + .await + .expect("command channel must be open"); + + // Drain handshake messages so the queue front is SubmitSharesExtended. + sniffer + .wait_for_message_type_and_clean_queue( + MessageDirection::ToUpstream, + MESSAGE_TYPE_OPEN_EXTENDED_MINING_CHANNEL, + ) + .await; + + sniffer + .wait_for_message_type( + MessageDirection::ToUpstream, + MESSAGE_TYPE_SUBMIT_SHARES_EXTENDED, + ) + .await; + + let msg = sniffer.next_message_from_downstream(); + match msg { + Some((_, AnyMessage::Mining(Mining::SubmitSharesExtended(m)))) => { + assert_eq!( + m.extranonce.inner_as_ref().len(), + 16, + "extranonce wire length must match miner alloc" + ); + } + other => panic!("expected SubmitSharesExtended, got {:?}", other), + } + + shutdown.cancel(); +} diff --git a/mujina-miner/src/stratum_v2/mod.rs b/mujina-miner/src/stratum_v2/mod.rs new file mode 100644 index 00000000..e2532267 --- /dev/null +++ b/mujina-miner/src/stratum_v2/mod.rs @@ -0,0 +1,43 @@ +//! Stratum V2 protocol client module. +//! +//! Provides a Noise-encrypted Stratum V2 client using Extended Channels. +//! Extended Channels supply `coinbase_tx_prefix`, `merkle_path`, and related +//! fields to the miner, enabling the existing `MerkleRootKind::Computed` path +//! without requiring ASIC-level `nVersion` bit rolling — see +//! [Mining Protocol > Extended Channel][sv2-ec]. +//! +//! Networking (Noise_NX handshake, encrypted frame I/O, DNS resolution, TCP +//! timeouts) is delegated to the [`stratum_apps`] library. +//! +//! # Architecture +//! +//! The [`StratumV2Client`] is a single-connection protocol session: +//! +//! 1. **DNS resolve** → TCP connect → Noise NX handshake (via +//! [`stratum_apps::network_helpers`]) +//! 2. **SetupConnection** — negotiate protocol version 2 +//! 3. **OpenExtendedMiningChannel** — establish the mining channel +//! 4. **Event loop** — `select!` over incoming frames, commands, and shutdown +//! +//! Read and write halves are decoupled: a spawned reader task does blocking +//! `read_frame()` calls (which are not cancellation-safe), forwarding decoded +//! messages to the main event loop via a channel. The main loop `select!`s +//! over the reader channel, command channel, and shutdown token. +//! +//! # References +//! +//! - [Protocol Security > URL Scheme and Pool Authority Key][sv2-url] +//! - [Mining Protocol > Extended Channel][sv2-ec] +//! +//! [sv2-url]: https://github.com/stratum-mining/sv2-spec/blob/main/04-Protocol-Security.md#47-url-scheme-and-pool-authority-key +//! [sv2-ec]: https://github.com/stratum-mining/sv2-spec/blob/main/05-Mining-Protocol.md#522-extended-channel + +mod client; +mod error; +#[cfg(test)] +mod integration_tests; + +pub use client::constants; +pub(crate) use client::target_from_le_bytes; +pub use client::{ClientCommand, ClientEvent, ClientOutcome, PoolConfig, StratumV2Client}; +pub use error::{StratumV2Error, StratumV2Result};