From a12781fe7114f784b2eedc525212c15281cc4a93 Mon Sep 17 00:00:00 2001 From: Raunak Kumar Date: Tue, 12 May 2026 21:22:16 +0000 Subject: [PATCH 1/4] Initial commit for share_info --- Cargo.lock | 119 ++++++- Cargo.toml | 5 + dummy.toml | 49 --- src/app.rs | 120 ++++++- src/components/mod.rs | 1 + src/components/p2pool_client.rs | 249 +++++++++++++- src/components/p2pool_status_view.rs | 304 ++++++++++++++++- src/components/p2pool_websocket.rs | 317 ++++++++++++++++++ src/main.rs | 3 + ...i__tests__p2pool_status_screen_render.snap | 3 +- 10 files changed, 1097 insertions(+), 73 deletions(-) delete mode 100644 dummy.toml create mode 100644 src/components/p2pool_websocket.rs diff --git a/Cargo.lock b/Cargo.lock index 124d572..582bffd 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -237,6 +237,12 @@ version = "1.25.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c8efb64bd706a16a1bdde310ae86b351e4d21550d98d056f22f8a7f7a2183fec" +[[package]] +name = "byteorder" +version = "1.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1fd0f2584146f6f2ef48085050886acf353beff7305ebd1ae69500e27c67f64b" + [[package]] name = "bytes" version = "1.11.1" @@ -509,6 +515,12 @@ dependencies = [ "syn 2.0.111", ] +[[package]] +name = "data-encoding" +version = "2.11.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a4ae5f15dda3c708c0ade84bfee31ccab44a3da4f88015ed22f63732abe300c8" + [[package]] name = "deltae" version = "0.3.2" @@ -782,6 +794,23 @@ dependencies = [ "futures-util", ] +[[package]] +name = "futures-io" +version = "0.3.32" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cecba35d7ad927e23624b22ad55235f2239cfa44fd10428eecbeba6d6a717718" + +[[package]] +name = "futures-macro" +version = "0.3.32" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e835b70203e41293343137df5c0664546da5745f82ec9b84d40be8336958447b" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.111", +] + [[package]] name = "futures-sink" version = "0.3.32" @@ -801,6 +830,9 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "389ca41296e6190b48053de0321d02a77f32f8a5d2461dd38762c0593805c6d6" dependencies = [ "futures-core", + "futures-io", + "futures-macro", + "futures-sink", "futures-task", "pin-project-lite", "slab", @@ -1028,7 +1060,7 @@ dependencies = [ "tokio", "tokio-rustls", "tower-service", - "webpki-roots", + "webpki-roots 1.0.7", ] [[package]] @@ -1693,10 +1725,12 @@ name = "pdm" version = "0.1.0" dependencies = [ "anyhow", + "base64 0.22.1", "bitcoin", "config 0.15.19", "crossterm", "directories", + "futures-util", "insta", "mockito", "p2poolv2_config", @@ -1707,9 +1741,11 @@ dependencies = [ "serial_test", "tempfile", "tokio", + "tokio-tungstenite", "toml 0.8.23", "toml_edit", "unicode-width", + "url", ] [[package]] @@ -1962,6 +1998,8 @@ version = "0.8.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "34af8d1a0e25924bc5b7c43c079c942339d8f0a8b57c39049bef581b46327404" dependencies = [ + "libc", + "rand_chacha 0.3.1", "rand_core 0.6.4", ] @@ -1971,10 +2009,20 @@ version = "0.9.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "44c5af06bb1b7d3216d91932aed5265164bf384dc89cd6ba05cf59a35f5f76ea" dependencies = [ - "rand_chacha", + "rand_chacha 0.9.0", "rand_core 0.9.5", ] +[[package]] +name = "rand_chacha" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e6c10a63a0fa32252be49d21e7709d4d4baf8d231c2dbce1eaa8141b9b127d88" +dependencies = [ + "ppv-lite86", + "rand_core 0.6.4", +] + [[package]] name = "rand_chacha" version = "0.9.0" @@ -1990,6 +2038,9 @@ name = "rand_core" version = "0.6.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ec0be4795e2f6a28069bec0b5ff3e2ac9bafc99e6a9a7dc3547996c5c816922c" +dependencies = [ + "getrandom 0.2.16", +] [[package]] name = "rand_core" @@ -2175,7 +2226,7 @@ dependencies = [ "wasm-bindgen", "wasm-bindgen-futures", "web-sys", - "webpki-roots", + "webpki-roots 1.0.7", ] [[package]] @@ -2503,6 +2554,17 @@ dependencies = [ "syn 2.0.111", ] +[[package]] +name = "sha1" +version = "0.10.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e3bf829a2d51ab4a5ddf1352d8470c140cadc8301b2ae1789db023f01cedd6ba" +dependencies = [ + "cfg-if", + "cpufeatures", + "digest", +] + [[package]] name = "sha2" version = "0.10.9" @@ -2929,6 +2991,22 @@ dependencies = [ "tokio", ] +[[package]] +name = "tokio-tungstenite" +version = "0.24.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "edc5f74e248dc973e0dbb7b74c7e0d6fcc301c694ff50049504004ef4d0cdcd9" +dependencies = [ + "futures-util", + "log", + "rustls", + "rustls-pki-types", + "tokio", + "tokio-rustls", + "tungstenite", + "webpki-roots 0.26.11", +] + [[package]] name = "tokio-util" version = "0.7.18" @@ -3135,6 +3213,26 @@ version = "0.2.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b" +[[package]] +name = "tungstenite" +version = "0.24.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "18e5b8366ee7a95b16d32197d0b2604b43a0be89dc5fac9f8e96ccafbaedda8a" +dependencies = [ + "byteorder", + "bytes", + "data-encoding", + "http", + "httparse", + "log", + "rand 0.8.5", + "rustls", + "rustls-pki-types", + "sha1", + "thiserror 1.0.69", + "utf-8", +] + [[package]] name = "typeid" version = "1.0.3" @@ -3206,6 +3304,12 @@ dependencies = [ "serde", ] +[[package]] +name = "utf-8" +version = "0.7.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "09cc8ee72d2a9becf2f2febe0205bbed8fc6615b7cb429ad062dc7b7ddd036a9" + [[package]] name = "utf8_iter" version = "1.0.4" @@ -3403,6 +3507,15 @@ dependencies = [ "wasm-bindgen", ] +[[package]] +name = "webpki-roots" +version = "0.26.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "521bc38abb08001b01866da9f51eb7c5d647a19260e00054a8c7fd5f9e57f7a9" +dependencies = [ + "webpki-roots 1.0.7", +] + [[package]] name = "webpki-roots" version = "1.0.7" diff --git a/Cargo.toml b/Cargo.toml index a1cd321..23c737e 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -19,6 +19,11 @@ bitcoin = "0.32.5" toml_edit = "0.22" reqwest = { version = "0.12", features = ["json", "rustls-tls"] } tokio = { version = "1", features = ["rt-multi-thread", "macros", "sync", "time"] } +base64 = "0.22.1" +futures-util = "0.3" +tokio-tungstenite = { version = "0.24", features = ["rustls-tls-webpki-roots"] } +url = "2" +serde_json = "1.0.135" [dev-dependencies] insta = "1.44.3" diff --git a/dummy.toml b/dummy.toml deleted file mode 100644 index bd5642b..0000000 --- a/dummy.toml +++ /dev/null @@ -1,49 +0,0 @@ - -[network] -listen_address = "/ip4/127.0.0.1/tcp/6884" -dial_peers = [] -max_pending_incoming = 10 -max_pending_outgoing = 10 -max_established_incoming = 50 -max_established_outgoing = 50 -max_established_per_peer = 1 -max_workbase_per_second = 10 -max_userworkbase_per_second = 10 -max_miningshare_per_second = 100 -max_inventory_per_second = 100 -max_transaction_per_second = 100 -max_requests_per_second = 100 -dial_timeout_secs = 30 - -[store] -path = "./store.db" -background_task_frequency_hours = 24 -pplns_ttl_days = 7 - -[stratum] -hostname = "pool.example.com" -port = 3333 -start_difficulty = 10000 -minimum_difficulty = 100 -solo_address = "tb1qyazxde6558qj6z3d9np5e6msmrspwpf6k0qggk" -bootstrap_address = "tb1qyazxde6558qj6z3d9np5e6msmrspwpf6k0qggk" -zmqpubhashblock = "tcp://127.0.0.1:28332" -network = "signet" -version_mask = "1fffe000" -difficulty_multiplier = 1.0 -pool_signature = "P2Poolv2" - -[bitcoinrpc] -url = "http://127.0.0.1:38332" -username = "p2pool" -password = "p2pool" - -[logging] -file = "./logs/p2pool.log" -console = true -level = "info" -stats_dir = "./logs/stats" - -[api] -hostname = "127.0.0.1" -port = 46884 diff --git a/src/app.rs b/src/app.rs index 3a6a6de..824b447 100644 --- a/src/app.rs +++ b/src/app.rs @@ -5,8 +5,11 @@ use crate::bitcoin_config::ConfigEntry as BitcoinEntry; use crate::components::bitcoin_config_view::BitcoinConfigView; use crate::components::file_explorer::FileExplorer; -use crate::components::p2pool_client::{ChainInfo, P2PoolClient, PeerInfo}; +use crate::components::p2pool_client::{ChainInfo, P2PoolClient, PeerInfo, SharesResponse}; use crate::components::p2pool_config_view::P2PoolConfigView; +use crate::components::p2pool_websocket::{ + LiveP2PoolEvent, LivePeerEvent, LiveShare, P2PoolWebSocketClient, +}; use crate::components::settings_view::SettingsView; use crate::settings::Settings; use p2poolv2_config::Config as P2PoolConfig; @@ -34,7 +37,7 @@ pub const BITCOIN_STATUS_TABS: &[&str] = &["Chain Info", "System", "Logs", "Peer pub const MAX_BITCOIN_STATUS_TAB: usize = BITCOIN_STATUS_TABS.len() - 1; /// Tab labels for the P2Pool Status view -pub const P2POOL_STATUS_TABS: &[&str] = &["Chain Info", "Peers Info"]; +pub const P2POOL_STATUS_TABS: &[&str] = &["Chain Info", "Shares", "Peers Info"]; pub const MAX_P2POOL_STATUS_TAB: usize = P2POOL_STATUS_TABS.len() - 1; @@ -104,6 +107,7 @@ pub struct App { pub bitcoin_status_tab: usize, pub settings: Settings, pub p2pool_client: P2PoolClient, + pub p2pool_websocket_client: P2PoolWebSocketClient, /// Cached value of the `HOME` environment variable, used for path display. /// Populated once at startup to avoid repeated syscalls during rendering. pub home_dir: String, @@ -113,12 +117,22 @@ pub struct App { pub p2pool_status_tab: usize, pub chain_info: Option, pub p2pool_chain_info_error: Option, + pub share_info: Option, + pub p2pool_share_info_error: Option, pub peer_info: Option>, pub p2pool_peer_info_error: Option, + pub live_shares: Vec, + pub live_peer_events: Vec, + pub p2pool_live_error: Option, + pub p2pool_live_stream_started: bool, + pub p2pool_live_tx: mpsc::UnboundedSender>, + pub p2pool_live_rx: mpsc::UnboundedReceiver>, // async channel to receive chain info updates from the background task that // fetches it when the P2Pool Status screen is opened. pub chain_info_tx: mpsc::UnboundedSender>, pub chain_info_rx: mpsc::UnboundedReceiver>, + pub share_info_tx: mpsc::UnboundedSender>, + pub share_info_rx: mpsc::UnboundedReceiver>, pub peer_info_tx: mpsc::UnboundedSender>>, pub peer_info_rx: mpsc::UnboundedReceiver>>, } @@ -128,6 +142,8 @@ impl App { pub fn new() -> App { let (chain_info_tx, chain_info_rx) = mpsc::unbounded_channel(); let (peer_info_tx, peer_info_rx) = mpsc::unbounded_channel(); + let (share_info_tx, share_info_rx) = mpsc::unbounded_channel(); + let (p2pool_live_tx, p2pool_live_rx) = mpsc::unbounded_channel(); App { current_screen: CurrentScreen::Home, sidebar_index: 0, @@ -143,15 +159,26 @@ impl App { bitcoin_status_tab: 0, settings: Settings::default(), p2pool_client: P2PoolClient::new(), + p2pool_websocket_client: P2PoolWebSocketClient::new(), home_dir: std::env::var("HOME").unwrap_or_default(), config_dir: crate::settings::config_dir().unwrap_or_default(), p2pool_status_tab: 0, chain_info: None, p2pool_chain_info_error: None, + share_info: None, + p2pool_share_info_error: None, peer_info: None, p2pool_peer_info_error: None, + live_shares: Vec::new(), + live_peer_events: Vec::new(), + p2pool_live_error: None, + p2pool_live_stream_started: false, + p2pool_live_tx, + p2pool_live_rx, chain_info_tx, chain_info_rx, + share_info_tx, + share_info_rx, peer_info_tx, peer_info_rx, } @@ -160,6 +187,7 @@ impl App { #[must_use] pub fn new_with_client(client: P2PoolClient) -> App { let mut app = App::new(); + app.p2pool_websocket_client = client.websocket_client(); app.p2pool_client = client; app } @@ -195,6 +223,72 @@ impl App { } } + pub fn poll_share_info(&mut self) { + while let Ok(result) = self.share_info_rx.try_recv() { + match result { + Ok(info) => { + self.share_info = Some(info); + self.p2pool_share_info_error = None; + } + Err(e) => { + self.share_info = None; + self.p2pool_share_info_error = Some(e.to_string()); + } + } + } + } + + pub fn poll_live_p2pool_events(&mut self) { + while let Ok(result) = self.p2pool_live_rx.try_recv() { + match result { + Ok(LiveP2PoolEvent::Share(share)) => { + Self::push_limited(&mut self.live_shares, share, 50); + self.p2pool_live_error = None; + } + Ok(LiveP2PoolEvent::Peer(peer_event)) => { + self.apply_live_peer_event(&peer_event); + Self::push_limited(&mut self.live_peer_events, peer_event, 50); + self.p2pool_live_error = None; + } + Err(e) => { + self.p2pool_live_error = Some(e.to_string()); + self.p2pool_live_stream_started = false; + } + } + } + } + + pub fn poll_live_shares(&mut self) { + self.poll_live_p2pool_events(); + } + + fn push_limited(items: &mut Vec, item: T, max_len: usize) { + items.push(item); + if items.len() > max_len { + let extra = items.len() - max_len; + items.drain(0..extra); + } + } + + fn apply_live_peer_event(&mut self, event: &LivePeerEvent) { + if event.status.eq_ignore_ascii_case("disconnected") { + if let Some(peers) = &mut self.peer_info { + peers.retain(|peer| peer.peer_id != event.peer_id); + } + return; + } + + let peers = self.peer_info.get_or_insert_with(Vec::new); + if let Some(peer) = peers.iter_mut().find(|peer| peer.peer_id == event.peer_id) { + peer.status = Some(event.status.clone()); + } else { + peers.push(PeerInfo { + peer_id: event.peer_id.clone(), + status: Some(event.status.clone()), + }); + } + } + // Logic to switch between sidebar items pub fn toggle_menu(&mut self) { if self.current_screen == CurrentScreen::BitcoinConfig { @@ -214,8 +308,13 @@ impl App { if self.current_screen == CurrentScreen::P2PoolStatus { let chain_client = self.p2pool_client.clone(); let chain_tx = self.chain_info_tx.clone(); + let share_client = self.p2pool_client.clone(); + let share_tx = self.share_info_tx.clone(); let peer_client = self.p2pool_client.clone(); let peer_tx = self.peer_info_tx.clone(); + let websocket_client = self.p2pool_websocket_client.clone(); + let live_tx = self.p2pool_live_tx.clone(); + let start_live_stream = !self.p2pool_live_stream_started; if let Ok(handle) = tokio::runtime::Handle::try_current() { handle.spawn(async move { @@ -223,10 +322,27 @@ impl App { let _ = chain_tx.send(res.map_err(anyhow::Error::from)); }); + handle.spawn(async move { + let res = share_client.fetch_recent_shares(10).await; + let _ = share_tx.send(res.map_err(anyhow::Error::from)); + }); + handle.spawn(async move { let res = peer_client.fetch_peer_info().await; let _ = peer_tx.send(res.map_err(anyhow::Error::from)); }); + + if start_live_stream { + self.p2pool_live_stream_started = true; + handle.spawn(async move { + if let Err(error) = websocket_client + .subscribe_live_events(live_tx.clone()) + .await + { + let _ = live_tx.send(Err(error)); + } + }); + } } } } diff --git a/src/components/mod.rs b/src/components/mod.rs index 1f94aa6..874a783 100644 --- a/src/components/mod.rs +++ b/src/components/mod.rs @@ -11,6 +11,7 @@ pub mod ln_status_view; pub mod p2pool_client; pub mod p2pool_config_view; pub mod p2pool_status_view; +pub mod p2pool_websocket; pub mod settings_view; pub mod shares_market_view; pub mod status_bar; diff --git a/src/components/p2pool_client.rs b/src/components/p2pool_client.rs index 85fe225..af47e35 100644 --- a/src/components/p2pool_client.rs +++ b/src/components/p2pool_client.rs @@ -2,17 +2,22 @@ // // SPDX-License-Identifier: AGPL-3.0-or-later +use crate::components::p2pool_websocket::P2PoolWebSocketClient; use crate::config::load_api_config; use reqwest::Client; use serde::Deserialize; +use serde::Deserializer; +use serde::de::DeserializeOwned; use std::time::Duration; const REQUEST_TIMEOUT_SECONDS: u64 = 10; +const TESTNET4_FALLBACK_BASE_URL: &str = "https://testnet4.p2poolv2.org"; #[derive(Debug, Clone)] pub struct P2PoolClient { client: Client, base_url: String, + fallback_base_url: Option, auth_credentials: Option<(String, String)>, } @@ -30,6 +35,36 @@ pub struct PeerInfo { pub status: Option, } +#[derive(Debug, Clone, Deserialize)] +pub struct ShareInfo { + pub blockhash: String, + pub prev_blockhash: String, + pub height: u64, + pub miner_address: String, + pub timestamp: u64, + #[serde(deserialize_with = "deserialize_string_or_number")] + pub bits: String, + #[serde(default)] + pub uncles: Vec, +} + +#[derive(Debug, Clone, Deserialize)] +pub struct UncleInfo { + pub blockhash: String, + pub prev_blockhash: String, + pub miner_address: String, + pub timestamp: u64, + pub height: u64, +} + +#[derive(Debug, Clone, Deserialize)] +pub struct SharesResponse { + pub from_height: u64, + pub to_height: u64, + #[serde(default)] + pub shares: Vec, +} + fn build_client() -> Client { Client::builder() .timeout(Duration::from_secs(REQUEST_TIMEOUT_SECONDS)) @@ -44,11 +79,15 @@ impl P2PoolClient { let credentials = cfg.auth_user.zip(cfg.auth_pass); (cfg.base_url, credentials) }) - .unwrap_or_else(|_| ("http://localhost:46884".to_string(), None)); + .unwrap_or_else(|_| (TESTNET4_FALLBACK_BASE_URL.to_string(), None)); + + let fallback_base_url = (base_url != TESTNET4_FALLBACK_BASE_URL) + .then(|| TESTNET4_FALLBACK_BASE_URL.to_string()); Self { client: build_client(), base_url, + fallback_base_url, auth_credentials, } } @@ -57,6 +96,7 @@ impl P2PoolClient { Self { client: build_client(), base_url: base_url.into(), + fallback_base_url: None, auth_credentials: None, } } @@ -65,6 +105,7 @@ impl P2PoolClient { Self { client, base_url: base_url.into(), + fallback_base_url: None, auth_credentials: None, } } @@ -74,32 +115,90 @@ impl P2PoolClient { self } - pub async fn fetch_chain_info(&self) -> Result { - let url = format!("{}/chain_info", self.base_url); - let mut request = self.client.get(url); + pub fn with_fallback_base_url(mut self, fallback_base_url: impl Into) -> Self { + self.fallback_base_url = Some(fallback_base_url.into()); + self + } + pub fn websocket_client(&self) -> P2PoolWebSocketClient { + let mut client = P2PoolWebSocketClient::with_base_url(self.base_url.clone()); if let Some((user, pass)) = &self.auth_credentials { - request = request.basic_auth(user, Some(pass)); + client = client.with_auth(user.clone(), pass.clone()); } + if let Some(fallback_base_url) = &self.fallback_base_url { + client = client.with_fallback_base_url(fallback_base_url.clone()); + } + client + } - let response = request.send().await?.error_for_status()?; - let data = response.json::().await?; - - Ok(data) + pub async fn fetch_chain_info(&self) -> Result { + self.fetch_json_with_fallback("/chain_info", &[]).await } pub async fn fetch_peer_info(&self) -> Result, reqwest::Error> { - let url = format!("{}/peers", self.base_url); + self.fetch_json_with_fallback("/peers", &[]).await + } + + pub async fn fetch_recent_shares(&self, num: u16) -> Result { + self.fetch_json_with_fallback("/shares", &[("num", num.min(100))]) + .await + } + + async fn fetch_json_with_fallback( + &self, + path: &str, + query: &[(&str, u16)], + ) -> Result + where + T: DeserializeOwned, + { + match self + .fetch_json_from_base_url(&self.base_url, path, query, true) + .await + { + Ok(data) => Ok(data), + Err(error) => { + if self.should_try_fallback(&error) { + if let Some(fallback_base_url) = &self.fallback_base_url { + return self + .fetch_json_from_base_url(fallback_base_url, path, query, false) + .await; + } + } + Err(error) + } + } + } + + async fn fetch_json_from_base_url( + &self, + base_url: &str, + path: &str, + query: &[(&str, u16)], + use_auth: bool, + ) -> Result + where + T: DeserializeOwned, + { + let url = format!("{}{}", base_url.trim_end_matches('/'), path); let mut request = self.client.get(url); - if let Some((user, pass)) = &self.auth_credentials { - request = request.basic_auth(user, Some(pass)); + if !query.is_empty() { + request = request.query(query); + } + + if use_auth { + if let Some((user, pass)) = &self.auth_credentials { + request = request.basic_auth(user, Some(pass)); + } } let response = request.send().await?.error_for_status()?; - let data = response.json::>().await?; + response.json::().await + } - Ok(data) + fn should_try_fallback(&self, error: &reqwest::Error) -> bool { + self.fallback_base_url.is_some() && (error.is_connect() || error.is_timeout()) } } @@ -109,12 +208,48 @@ impl Default for P2PoolClient { } } +fn deserialize_string_or_number<'de, D>(deserializer: D) -> Result +where + D: Deserializer<'de>, +{ + #[derive(Deserialize)] + #[serde(untagged)] + enum StringOrNumber { + String(String), + Number(u64), + } + + match StringOrNumber::deserialize(deserializer)? { + StringOrNumber::String(value) => Ok(value), + StringOrNumber::Number(value) => Ok(value.to_string()), + } +} + #[cfg(test)] mod tests { use super::*; - use mockito::Server; + use mockito::{Matcher, Server}; use serde_json::json; + #[test] + fn explicit_base_url_does_not_enable_network_fallback() { + let client = P2PoolClient::with_base_url("http://127.0.0.1:46884"); + + assert_eq!(client.base_url, "http://127.0.0.1:46884"); + assert_eq!(client.fallback_base_url, None); + } + + #[test] + fn fallback_base_url_can_be_configured() { + let client = P2PoolClient::with_base_url("http://127.0.0.1:46884") + .with_fallback_base_url("https://testnet4.p2poolv2.org"); + + assert_eq!( + client.fallback_base_url.as_deref(), + Some("https://testnet4.p2poolv2.org") + ); + } + #[tokio::test] async fn test_fetch_chain_info_success() { let mut server = Server::new_async().await; @@ -239,6 +374,90 @@ mod tests { mock.assert(); } + #[tokio::test] + async fn test_fetch_recent_shares_success() { + let mut server = Server::new_async().await; + + let mock = server + .mock("GET", "/shares") + .match_query(Matcher::UrlEncoded("num".into(), "2".into())) + .with_status(200) + .with_header("content-type", "application/json") + .with_body( + json!({ + "from_height": 41, + "to_height": 42, + "shares": [ + { + "blockhash": "0000share", + "prev_blockhash": "ffffprev", + "height": 42, + "miner_address": "miner-address", + "timestamp": 1700000000u64, + "bits": "1d00ffff", + "uncles": [ + { + "blockhash": "0000uncle", + "prev_blockhash": "ffffuncleprev", + "miner_address": "uncle-miner", + "timestamp": 1699999999u64, + "height": 41 + } + ] + } + ] + }) + .to_string(), + ) + .create(); + + let client = P2PoolClient::with_base_url(server.url()); + let result = client.fetch_recent_shares(2).await.unwrap(); + + assert_eq!(result.from_height, 41); + assert_eq!(result.to_height, 42); + assert_eq!(result.shares.len(), 1); + assert_eq!(result.shares[0].height, 42); + assert_eq!(result.shares[0].uncles.len(), 1); + mock.assert(); + } + + #[tokio::test] + async fn test_fetch_recent_shares_accepts_numeric_bits() { + let mut server = Server::new_async().await; + + let mock = server + .mock("GET", "/shares") + .match_query(Matcher::UrlEncoded("num".into(), "1".into())) + .with_status(200) + .with_header("content-type", "application/json") + .with_body( + json!({ + "from_height": 42, + "to_height": 42, + "shares": [ + { + "blockhash": "0000share", + "prev_blockhash": "ffffprev", + "height": 42, + "miner_address": "miner-address", + "timestamp": 1700000000u64, + "bits": 454130449, + "uncles": [] + } + ] + }) + .to_string(), + ) + .create(); + + let client = P2PoolClient::with_base_url(server.url()); + let result = client.fetch_recent_shares(1).await.unwrap(); + + assert_eq!(result.shares[0].bits, "454130449"); + mock.assert(); + } + #[tokio::test] async fn test_fetch_chain_info_errors_on_http_500() { let mut server = Server::new_async().await; diff --git a/src/components/p2pool_status_view.rs b/src/components/p2pool_status_view.rs index d249375..9419de0 100644 --- a/src/components/p2pool_status_view.rs +++ b/src/components/p2pool_status_view.rs @@ -5,12 +5,23 @@ use crate::app::{App, P2POOL_STATUS_TABS}; use ratatui::{ prelude::*, - widgets::{Block, Borders, Paragraph, Tabs, Wrap}, + widgets::{Block, Borders, Cell, Paragraph, Row, Table, Tabs, Wrap}, }; +use std::collections::HashSet; #[derive(Debug, Clone)] pub struct P2PoolStatusView; +#[derive(Debug)] +struct ShareTableEntry { + height: u64, + blockhash: String, + miner: String, + bits: String, + timestamp: u64, + uncles: usize, +} + impl P2PoolStatusView { #[must_use] pub fn new() -> Self { @@ -36,7 +47,8 @@ impl P2PoolStatusView { match app.p2pool_status_tab { 0 => Self::render_chain_info(f, app, outer[1]), - 1 => Self::render_peer_info(f, app, outer[1]), + 1 => Self::render_share_info(f, app, outer[1]), + 2 => Self::render_peer_info(f, app, outer[1]), _ => {} } } @@ -78,8 +90,46 @@ impl P2PoolStatusView { f.render_widget(paragraph, area); } + fn render_share_info(f: &mut Frame, app: &App, area: Rect) { + let mut rows = Self::share_rows(app); + if rows.is_empty() { + rows.push(Self::message_row(Self::share_empty_message(app))); + } + + let header = Row::new([ + "Height", + "Blockhash", + "Miner", + "Difficulty", + "Time", + "Uncles", + ]) + .style( + Style::default() + .fg(Color::Gray) + .add_modifier(Modifier::BOLD), + ) + .bottom_margin(1); + let widths = [ + Constraint::Length(7), + Constraint::Length(10), + Constraint::Length(10), + Constraint::Length(10), + Constraint::Min(12), + Constraint::Length(6), + ]; + + let table = Table::new(rows, widths) + .block(Block::default().borders(Borders::ALL).title(" Shares ")) + .header(header) + .column_spacing(1) + .style(Style::default().fg(Color::White)); + + f.render_widget(table, area); + } + fn render_peer_info(f: &mut Frame, app: &App, area: Rect) { - let text = if let Some(peers) = &app.peer_info { + let mut text = if let Some(peers) = &app.peer_info { if peers.is_empty() { vec![Line::from(Span::styled( "No connected peers", @@ -115,12 +165,260 @@ impl P2PoolStatusView { ))] }; + if let Some(err) = &app.p2pool_live_error { + text.push(Line::from("")); + text.push(Line::from(Span::styled( + format!("Live stream error: {err}"), + Style::default().fg(Color::Red), + ))); + } + + if !app.live_peer_events.is_empty() { + text.push(Line::from("")); + text.push(Line::from(Span::styled( + "Live Peer Events", + Style::default().add_modifier(Modifier::BOLD), + ))); + for event in app.live_peer_events.iter().rev().take(8) { + text.push(Line::from(format!( + "{}: {}", + event.status, + Self::short_value(&event.peer_id, 42) + ))); + } + } + let paragraph = Paragraph::new(text) .block(Block::default().borders(Borders::ALL).title(" Peers Info ")) .wrap(Wrap { trim: true }); f.render_widget(paragraph, area); } + + fn short_value(value: &str, max_len: usize) -> String { + if value.len() <= max_len { + return value.to_string(); + } + + if max_len <= 3 { + return value.chars().take(max_len).collect(); + } + + let head_len = (max_len - 3) / 2; + let tail_len = max_len - 3 - head_len; + let head: String = value.chars().take(head_len).collect(); + let tail: String = value + .chars() + .rev() + .take(tail_len) + .collect::>() + .into_iter() + .rev() + .collect(); + format!("{head}...{tail}") + } + + fn share_rows(app: &App) -> Vec> { + let mut entries = Vec::new(); + let mut seen = HashSet::new(); + + for share in app.live_shares.iter().rev() { + seen.insert(share.blockhash.clone()); + entries.push(ShareTableEntry { + height: share.height, + blockhash: share.blockhash.clone(), + miner: share.miner_address.clone(), + bits: share.bits.clone(), + timestamp: share.timestamp, + uncles: share.uncles.len(), + }); + } + + if let Some(info) = &app.share_info { + for share in info.shares.iter().rev() { + if seen.insert(share.blockhash.clone()) { + entries.push(ShareTableEntry { + height: share.height, + blockhash: share.blockhash.clone(), + miner: share.miner_address.clone(), + bits: share.bits.clone(), + timestamp: share.timestamp, + uncles: share.uncles.len(), + }); + } + } + } + + entries.sort_by(|left, right| { + right + .height + .cmp(&left.height) + .then_with(|| right.timestamp.cmp(&left.timestamp)) + }); + + entries + .into_iter() + .take(50) + .map(|entry| { + Self::share_row( + entry.height, + &entry.blockhash, + &entry.miner, + &entry.bits, + entry.timestamp, + entry.uncles, + ) + }) + .collect() + } + + fn share_row( + height: u64, + blockhash: &str, + miner: &str, + bits: &str, + timestamp: u64, + uncles: usize, + ) -> Row<'static> { + Row::new(vec![ + Cell::from(height.to_string()), + Self::chip(Self::short_value(blockhash, 10)), + Self::chip(Self::short_value(miner, 10)), + Cell::from(Self::format_difficulty(bits)), + Cell::from(Self::format_timestamp(timestamp)), + Cell::from(uncles.to_string()), + ]) + .height(1) + } + + fn message_row(message: String) -> Row<'static> { + Row::new(vec![ + Cell::from(Span::styled(message, Style::default().fg(Color::DarkGray))), + Cell::from(""), + Cell::from(""), + Cell::from(""), + Cell::from(""), + Cell::from(""), + ]) + } + + fn share_empty_message(app: &App) -> String { + if let Some(err) = &app.p2pool_share_info_error { + return format!("Recent shares unavailable: {}", Self::short_value(err, 64)); + } + + if let Some(err) = &app.p2pool_live_error { + return format!("Live shares unavailable: {}", Self::short_value(err, 64)); + } + + "Waiting for share data...".to_string() + } + + fn chip(value: String) -> Cell<'static> { + Cell::from(Span::styled( + value, + Style::default().fg(Color::Gray).bg(Color::Black), + )) + } + + fn format_difficulty(bits: &str) -> String { + let Some(bits) = Self::parse_bits(bits) else { + return Self::short_value(bits, 10); + }; + + let exponent = (bits >> 24) as i32; + let mantissa = bits & 0x00ff_ffff; + if mantissa == 0 { + return "-".to_string(); + } + + let difficulty = (0x00ff_ff_u32 as f64 / mantissa as f64) * 256_f64.powi(0x1d - exponent); + if !difficulty.is_finite() || difficulty <= 0.0 { + return "-".to_string(); + } + + if difficulty >= 100.0 { + return Self::format_integer_with_commas(difficulty.round() as u64); + } + + if difficulty >= 1.0 { + return format!("{difficulty:.2}"); + } + + format!("{difficulty:.4}") + } + + fn parse_bits(bits: &str) -> Option { + let value = bits.trim(); + if value.is_empty() { + return None; + } + + if let Some(hex) = value + .strip_prefix("0x") + .or_else(|| value.strip_prefix("0X")) + { + return u32::from_str_radix(hex, 16).ok(); + } + + if value + .chars() + .any(|c| c.is_ascii_hexdigit() && c.is_ascii_alphabetic()) + { + return u32::from_str_radix(value, 16).ok(); + } + + value.parse::().ok() + } + + fn format_integer_with_commas(value: u64) -> String { + let digits = value.to_string(); + let mut formatted = String::with_capacity(digits.len() + digits.len() / 3); + for (index, digit) in digits.chars().rev().enumerate() { + if index > 0 && index % 3 == 0 { + formatted.push(','); + } + formatted.push(digit); + } + formatted.chars().rev().collect() + } + + fn format_timestamp(timestamp: u64) -> String { + let timestamp = if timestamp > 10_000_000_000 { + timestamp / 1_000 + } else { + timestamp + }; + let days = (timestamp / 86_400) as i64; + let seconds = timestamp % 86_400; + let hour = seconds / 3_600; + let minute = (seconds % 3_600) / 60; + let second = seconds % 60; + let (year, month, day) = Self::civil_from_days(days); + let suffix = if hour < 12 { "AM" } else { "PM" }; + let hour = match hour % 12 { + 0 => 12, + value => value, + }; + + format!("{month}/{day}/{year}, {hour}:{minute:02}:{second:02} {suffix}") + } + + fn civil_from_days(days_since_epoch: i64) -> (i32, u32, u32) { + let days = days_since_epoch + 719_468; + let era = if days >= 0 { days } else { days - 146_096 } / 146_097; + let day_of_era = days - era * 146_097; + let year_of_era = + (day_of_era - day_of_era / 1_460 + day_of_era / 36_524 - day_of_era / 146_096) / 365; + let year = year_of_era + era * 400; + let day_of_year = day_of_era - (365 * year_of_era + year_of_era / 4 - year_of_era / 100); + let month_param = (5 * day_of_year + 2) / 153; + let day = day_of_year - (153 * month_param + 2) / 5 + 1; + let month = month_param + if month_param < 10 { 3 } else { -9 }; + let year = year + if month <= 2 { 1 } else { 0 }; + + (year as i32, month as u32, day as u32) + } } impl Default for P2PoolStatusView { diff --git a/src/components/p2pool_websocket.rs b/src/components/p2pool_websocket.rs new file mode 100644 index 0000000..9481ab8 --- /dev/null +++ b/src/components/p2pool_websocket.rs @@ -0,0 +1,317 @@ +// SPDX-FileCopyrightText: 2024 PDM Authors +// +// SPDX-License-Identifier: AGPL-3.0-or-later + +use crate::config::load_api_config; +use anyhow::{Context, Result}; +use base64::Engine as _; +use base64::engine::general_purpose::STANDARD; +use futures_util::{SinkExt, StreamExt}; +use serde::{Deserialize, Deserializer}; +use tokio::sync::mpsc; +use tokio_tungstenite::connect_async; +use tokio_tungstenite::tungstenite::Message; +use url::Url; + +const TESTNET4_FALLBACK_BASE_URL: &str = "https://testnet4.p2poolv2.org"; + +#[derive(Debug, Clone)] +pub struct P2PoolWebSocketClient { + base_url: String, + fallback_base_url: Option, + auth_credentials: Option<(String, String)>, +} + +#[derive(Debug, Clone, Deserialize, PartialEq, Eq)] +pub struct ShareEventData { + pub blockhash: String, + pub prev_blockhash: String, + pub height: u64, + pub miner_address: String, + pub timestamp: u64, + #[serde(deserialize_with = "deserialize_string_or_number")] + pub bits: String, + #[serde(default)] + pub uncles: Vec, +} + +#[derive(Debug, Clone, Deserialize, PartialEq, Eq)] +pub struct PeerEventData { + pub peer_id: String, + pub status: String, +} + +#[derive(Debug, Clone, Deserialize, PartialEq, Eq)] +#[serde(tag = "topic", content = "data")] +pub enum WebSocketEvent { + #[serde(rename = "Share")] + Share(ShareEventData), + #[serde(rename = "Peer")] + Peer(PeerEventData), +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct LiveShare { + pub blockhash: String, + pub prev_blockhash: String, + pub height: u64, + pub miner_address: String, + pub timestamp: u64, + pub bits: String, + pub uncles: Vec, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct LivePeerEvent { + pub peer_id: String, + pub status: String, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum LiveP2PoolEvent { + Share(LiveShare), + Peer(LivePeerEvent), +} + +impl P2PoolWebSocketClient { + pub fn new() -> Self { + let (base_url, auth_credentials) = load_api_config() + .map(|cfg| { + let credentials = cfg.auth_user.zip(cfg.auth_pass); + (cfg.base_url, credentials) + }) + .unwrap_or_else(|_| (TESTNET4_FALLBACK_BASE_URL.to_string(), None)); + + let fallback_base_url = (base_url != TESTNET4_FALLBACK_BASE_URL) + .then(|| TESTNET4_FALLBACK_BASE_URL.to_string()); + + Self { + base_url, + fallback_base_url, + auth_credentials, + } + } + + pub fn with_base_url(base_url: impl Into) -> Self { + Self { + base_url: base_url.into(), + fallback_base_url: None, + auth_credentials: None, + } + } + + pub fn with_auth(mut self, user: String, pass: String) -> Self { + self.auth_credentials = Some((user, pass)); + self + } + + pub fn with_fallback_base_url(mut self, fallback_base_url: impl Into) -> Self { + self.fallback_base_url = Some(fallback_base_url.into()); + self + } + + fn ws_url(&self, path: &str) -> Result { + self.ws_url_from_base_url(&self.base_url, path) + } + + fn ws_url_from_base_url(&self, base_url: &str, path: &str) -> Result { + let mut url = Url::parse(base_url) + .with_context(|| format!("Failed to parse base URL: {base_url}"))?; + + match url.scheme() { + "http" => url.set_scheme("ws").unwrap(), + "https" => url.set_scheme("wss").unwrap(), + _ => {} + } + + url.set_path(path); + Ok(url) + } + + fn ws_url_with_auth(&self, path: &str) -> Result { + let mut url = self.ws_url(path)?; + if let Some((user, pass)) = &self.auth_credentials { + let token = STANDARD.encode(format!("{}:{}", user, pass)); + url.query_pairs_mut().append_pair("token", &token); + } + Ok(url) + } + + pub async fn subscribe_live_events( + &self, + tx: mpsc::UnboundedSender>, + ) -> anyhow::Result<()> { + let url = self.ws_url_with_auth("/ws")?; + match self.subscribe_live_events_at(url, tx.clone()).await { + Ok(()) => Ok(()), + Err(primary_error) => { + if let Some(fallback_base_url) = &self.fallback_base_url { + let fallback_url = self.ws_url_from_base_url(fallback_base_url, "/ws")?; + if self + .subscribe_live_events_at(fallback_url, tx) + .await + .is_ok() + { + return Ok(()); + } + } + Err(primary_error) + } + } + } + + async fn subscribe_live_events_at( + &self, + url: Url, + tx: mpsc::UnboundedSender>, + ) -> anyhow::Result<()> { + let (stream, _) = connect_async(url.as_str()).await?; + let (mut write, mut read) = stream.split(); + + for topic in ["shares", "peers"] { + let subscribe_message = serde_json::json!({ + "action": "subscribe", + "topic": topic, + }) + .to_string(); + write.send(Message::Text(subscribe_message)).await?; + } + + while let Some(message_result) = read.next().await { + let message = message_result?; + if let Message::Text(text) = message { + match serde_json::from_str::(&text) { + Ok(WebSocketEvent::Share(data)) => { + let live_share = LiveShare { + blockhash: data.blockhash, + prev_blockhash: data.prev_blockhash, + height: data.height, + miner_address: data.miner_address, + timestamp: data.timestamp, + bits: data.bits, + uncles: data.uncles, + }; + let _ = tx.send(Ok(LiveP2PoolEvent::Share(live_share))); + } + Ok(WebSocketEvent::Peer(data)) => { + let live_peer = LivePeerEvent { + peer_id: data.peer_id, + status: data.status, + }; + let _ = tx.send(Ok(LiveP2PoolEvent::Peer(live_peer))); + } + Err(error) => { + let _ = tx.send(Err(anyhow::Error::new(error))); + } + } + } + } + + Ok(()) + } +} + +impl Default for P2PoolWebSocketClient { + fn default() -> Self { + Self::new() + } +} + +fn deserialize_string_or_number<'de, D>(deserializer: D) -> Result +where + D: Deserializer<'de>, +{ + #[derive(Deserialize)] + #[serde(untagged)] + enum StringOrNumber { + String(String), + Number(u64), + } + + match StringOrNumber::deserialize(deserializer)? { + StringOrNumber::String(value) => Ok(value), + StringOrNumber::Number(value) => Ok(value.to_string()), + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn ws_url_converts_http_to_ws_and_encodes_auth_token() { + let client = P2PoolWebSocketClient::with_base_url("http://127.0.0.1:46884") + .with_auth("user".into(), "password".into()); + + let url = client.ws_url_with_auth("/ws").unwrap(); + + assert_eq!( + url.as_str(), + "ws://127.0.0.1:46884/ws?token=dXNlcjpwYXNzd29yZA%3D%3D" + ); + } + + #[test] + fn ws_url_converts_https_fallback_to_wss() { + let client = P2PoolWebSocketClient::with_base_url("https://testnet4.p2poolv2.org"); + + let url = client.ws_url("/ws").unwrap(); + + assert_eq!(url.as_str(), "wss://testnet4.p2poolv2.org/ws"); + } + + #[test] + fn websocket_event_accepts_share_messages() { + let event: WebSocketEvent = serde_json::from_value(serde_json::json!({ + "topic": "Share", + "data": { + "blockhash": "0000", + "prev_blockhash": "ffff", + "height": 42, + "miner_address": "miner", + "timestamp": 1700000000, + "bits": "1d00ffff", + "uncles": ["aaaa"] + } + })) + .unwrap(); + + assert!(matches!(event, WebSocketEvent::Share(_))); + } + + #[test] + fn websocket_event_accepts_numeric_bits() { + let event: WebSocketEvent = serde_json::from_value(serde_json::json!({ + "topic": "Share", + "data": { + "blockhash": "0000", + "prev_blockhash": "ffff", + "height": 42, + "miner_address": "miner", + "timestamp": 1700000000, + "bits": 454130449, + "uncles": [] + } + })) + .unwrap(); + + let WebSocketEvent::Share(data) = event else { + panic!("expected share event"); + }; + assert_eq!(data.bits, "454130449"); + } + + #[test] + fn websocket_event_accepts_peer_messages() { + let event: WebSocketEvent = serde_json::from_value(serde_json::json!({ + "topic": "Peer", + "data": { + "peer_id": "12D3KooWPeerOne", + "status": "Connected" + } + })) + .unwrap(); + + assert!(matches!(event, WebSocketEvent::Peer(_))); + } +} diff --git a/src/main.rs b/src/main.rs index 54125af..4053932 100644 --- a/src/main.rs +++ b/src/main.rs @@ -23,6 +23,7 @@ use crossterm::{ terminal::{EnterAlternateScreen, LeaveAlternateScreen, disable_raw_mode, enable_raw_mode}, }; use ratatui::{Terminal, backend::Backend, backend::CrosstermBackend}; +use std::{io, time::Duration}; use std::io; use std::time::Duration; @@ -83,7 +84,9 @@ where { loop { app.poll_chain_info(); + app.poll_share_info(); app.poll_peer_info(); + app.poll_live_p2pool_events(); terminal.draw(|f| ui::ui(f, app))?; if event::poll(Duration::from_millis(100))? { diff --git a/src/snapshots/pdm__ui__tests__p2pool_status_screen_render.snap b/src/snapshots/pdm__ui__tests__p2pool_status_screen_render.snap index 9614b70..3f096dd 100644 --- a/src/snapshots/pdm__ui__tests__p2pool_status_screen_render.snap +++ b/src/snapshots/pdm__ui__tests__p2pool_status_screen_render.snap @@ -1,5 +1,6 @@ --- source: src/ui.rs +assertion_line: 219 expression: terminal.backend() --- TestBackend { @@ -7,7 +8,7 @@ TestBackend { area: Rect { x: 0, y: 0, width: 80, height: 24 }, content: [ "┌ PDM ──────────────────┐┌ Info ───────────────────────────────────────────────┐", - "│Home ││ Chain Info │ Peers Info │", + "│Home ││ Chain Info │ Shares │ Peers Info │", "│Bitcoin Config ││ │", "│Bitcoin Status │└─────────────────────────────────────────────────────┘", "│P2Pool Config │┌ Chain Info ─────────────────────────────────────────┐", From 06e673eaea58efc9530663f389d3ece83e3a912b Mon Sep 17 00:00:00 2001 From: Raunak Kumar Date: Sun, 24 May 2026 01:08:08 +0000 Subject: [PATCH 2/4] move p2pool client urls to config and localize tests --- config/config.toml | 2 ++ src/components/p2pool_client.rs | 56 ++++++++++++++++++------------ src/components/p2pool_websocket.rs | 34 +++++++++--------- src/config.rs | 32 +++++++++++++++-- 4 files changed, 81 insertions(+), 43 deletions(-) diff --git a/config/config.toml b/config/config.toml index 8d74355..6ec999a 100644 --- a/config/config.toml +++ b/config/config.toml @@ -1,4 +1,6 @@ [api] +base_url = "http://127.0.0.1:46884" +# fallback_base_url = "http://127.0.0.1:46885" host = "127.0.0.1" port = 46884 auth_user = "p2pool" diff --git a/src/components/p2pool_client.rs b/src/components/p2pool_client.rs index af47e35..d063760 100644 --- a/src/components/p2pool_client.rs +++ b/src/components/p2pool_client.rs @@ -3,7 +3,7 @@ // SPDX-License-Identifier: AGPL-3.0-or-later use crate::components::p2pool_websocket::P2PoolWebSocketClient; -use crate::config::load_api_config; +use crate::config::{ApiConfig, load_api_config}; use reqwest::Client; use serde::Deserialize; use serde::Deserializer; @@ -11,7 +11,6 @@ use serde::de::DeserializeOwned; use std::time::Duration; const REQUEST_TIMEOUT_SECONDS: u64 = 10; -const TESTNET4_FALLBACK_BASE_URL: &str = "https://testnet4.p2poolv2.org"; #[derive(Debug, Clone)] pub struct P2PoolClient { @@ -74,21 +73,22 @@ fn build_client() -> Client { impl P2PoolClient { pub fn new() -> Self { - let (base_url, auth_credentials) = load_api_config() - .map(|cfg| { - let credentials = cfg.auth_user.zip(cfg.auth_pass); - (cfg.base_url, credentials) - }) - .unwrap_or_else(|_| (TESTNET4_FALLBACK_BASE_URL.to_string(), None)); + Self::from_config(load_api_config().unwrap_or_default()) + } - let fallback_base_url = (base_url != TESTNET4_FALLBACK_BASE_URL) - .then(|| TESTNET4_FALLBACK_BASE_URL.to_string()); + fn from_config(config: ApiConfig) -> Self { + let client = P2PoolClient::with_base_url(&config.base_url); - Self { - client: build_client(), - base_url, - fallback_base_url, - auth_credentials, + let client = if let Some(fallback) = &config.fallback_base_url { + client.with_fallback_base_url(fallback) + } else { + client + }; + + if let Some((user, pass)) = config.auth_user.zip(config.auth_pass) { + client.with_auth(user, pass) + } else { + client } } @@ -231,23 +231,33 @@ mod tests { use mockito::{Matcher, Server}; use serde_json::json; + const PRIMARY_BASE_URL: &str = "http://127.0.0.1:46884"; + const FALLBACK_BASE_URL: &str = "http://127.0.0.1:46885"; + + fn api_config(fallback_base_url: Option<&str>) -> ApiConfig { + ApiConfig { + base_url: PRIMARY_BASE_URL.to_string(), + fallback_base_url: fallback_base_url.map(str::to_string), + auth_user: None, + auth_pass: None, + } + } + #[test] fn explicit_base_url_does_not_enable_network_fallback() { - let client = P2PoolClient::with_base_url("http://127.0.0.1:46884"); + let config = api_config(None); + let client = P2PoolClient::from_config(config); - assert_eq!(client.base_url, "http://127.0.0.1:46884"); + assert_eq!(client.base_url, PRIMARY_BASE_URL); assert_eq!(client.fallback_base_url, None); } #[test] fn fallback_base_url_can_be_configured() { - let client = P2PoolClient::with_base_url("http://127.0.0.1:46884") - .with_fallback_base_url("https://testnet4.p2poolv2.org"); + let config = api_config(Some(FALLBACK_BASE_URL)); + let client = P2PoolClient::from_config(config); - assert_eq!( - client.fallback_base_url.as_deref(), - Some("https://testnet4.p2poolv2.org") - ); + assert_eq!(client.fallback_base_url.as_deref(), Some(FALLBACK_BASE_URL)); } #[tokio::test] diff --git a/src/components/p2pool_websocket.rs b/src/components/p2pool_websocket.rs index 9481ab8..fa3f967 100644 --- a/src/components/p2pool_websocket.rs +++ b/src/components/p2pool_websocket.rs @@ -2,7 +2,7 @@ // // SPDX-License-Identifier: AGPL-3.0-or-later -use crate::config::load_api_config; +use crate::config::{ApiConfig, load_api_config}; use anyhow::{Context, Result}; use base64::Engine as _; use base64::engine::general_purpose::STANDARD; @@ -13,8 +13,6 @@ use tokio_tungstenite::connect_async; use tokio_tungstenite::tungstenite::Message; use url::Url; -const TESTNET4_FALLBACK_BASE_URL: &str = "https://testnet4.p2poolv2.org"; - #[derive(Debug, Clone)] pub struct P2PoolWebSocketClient { base_url: String, @@ -75,20 +73,22 @@ pub enum LiveP2PoolEvent { impl P2PoolWebSocketClient { pub fn new() -> Self { - let (base_url, auth_credentials) = load_api_config() - .map(|cfg| { - let credentials = cfg.auth_user.zip(cfg.auth_pass); - (cfg.base_url, credentials) - }) - .unwrap_or_else(|_| (TESTNET4_FALLBACK_BASE_URL.to_string(), None)); + Self::from_config(load_api_config().unwrap_or_default()) + } - let fallback_base_url = (base_url != TESTNET4_FALLBACK_BASE_URL) - .then(|| TESTNET4_FALLBACK_BASE_URL.to_string()); + fn from_config(config: ApiConfig) -> Self { + let client = P2PoolWebSocketClient::with_base_url(&config.base_url); - Self { - base_url, - fallback_base_url, - auth_credentials, + let client = if let Some(fallback) = &config.fallback_base_url { + client.with_fallback_base_url(fallback) + } else { + client + }; + + if let Some((user, pass)) = config.auth_user.zip(config.auth_pass) { + client.with_auth(user, pass) + } else { + client } } @@ -253,11 +253,11 @@ mod tests { #[test] fn ws_url_converts_https_fallback_to_wss() { - let client = P2PoolWebSocketClient::with_base_url("https://testnet4.p2poolv2.org"); + let client = P2PoolWebSocketClient::with_base_url("https://127.0.0.1:46884"); let url = client.ws_url("/ws").unwrap(); - assert_eq!(url.as_str(), "wss://testnet4.p2poolv2.org/ws"); + assert_eq!(url.as_str(), "wss://127.0.0.1:46884/ws"); } #[test] diff --git a/src/config.rs b/src/config.rs index c7e5966..a70f086 100644 --- a/src/config.rs +++ b/src/config.rs @@ -5,12 +5,28 @@ use anyhow::Result; use config::{Config, File}; +const DEFAULT_API_HOST: &str = "127.0.0.1"; +const DEFAULT_API_PORT: u16 = 9332; + +#[derive(Debug, Clone)] pub struct ApiConfig { pub base_url: String, + pub fallback_base_url: Option, pub auth_user: Option, pub auth_pass: Option, } +impl Default for ApiConfig { + fn default() -> Self { + Self { + base_url: format!("http://{}:{}", DEFAULT_API_HOST, DEFAULT_API_PORT), + fallback_base_url: None, + auth_user: None, + auth_pass: None, + } + } +} + pub fn load_api_config() -> Result { let settings = Config::builder() .add_source(File::with_name("config/config").required(false)) @@ -19,13 +35,23 @@ pub fn load_api_config() -> Result { ) .build()?; - let host: String = settings.get("api.host").unwrap_or("127.0.0.1".into()); - let port: u16 = settings.get("api.port").unwrap_or(46884); + let host: String = settings + .get("api.host") + .unwrap_or_else(|_| DEFAULT_API_HOST.to_string()); + let port: u16 = settings.get("api.port").unwrap_or(DEFAULT_API_PORT); + let base_url: String = settings + .get("api.base_url") + .unwrap_or_else(|_| format!("http://{}:{}", host, port)); + let fallback_base_url: Option = settings + .get("api.fallback_base_url") + .ok() + .filter(|url: &String| !url.trim().is_empty()); let auth_user: Option = settings.get("api.auth_user").ok(); let auth_pass: Option = settings.get("api.auth_pass").ok(); Ok(ApiConfig { - base_url: format!("http://{}:{}", host, port), + base_url, + fallback_base_url, auth_user, auth_pass, }) From e418f819d187410b98ba9633ab21c8ec5475ee94 Mon Sep 17 00:00:00 2001 From: Raunak Kumar Date: Fri, 3 Jul 2026 17:53:26 +0000 Subject: [PATCH 3/4] test: Improve websocket coverage, add TLS fallback for redirects, fix clippy warnings --- Cargo.lock | 7 - src/components/file_explorer.rs | 2 +- src/components/p2pool_client.rs | 18 +-- src/components/p2pool_config_view.rs | 42 +++-- src/components/p2pool_status_view.rs | 222 +++++++++++++++++++++++++-- src/components/p2pool_websocket.rs | 212 +++++++++++++++++++++++-- src/config.rs | 2 +- src/main.rs | 50 +++--- src/p2poolv2_config.rs | 5 +- 9 files changed, 469 insertions(+), 91 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 582bffd..a45e3da 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -794,12 +794,6 @@ dependencies = [ "futures-util", ] -[[package]] -name = "futures-io" -version = "0.3.32" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "cecba35d7ad927e23624b22ad55235f2239cfa44fd10428eecbeba6d6a717718" - [[package]] name = "futures-macro" version = "0.3.32" @@ -830,7 +824,6 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "389ca41296e6190b48053de0321d02a77f32f8a5d2461dd38762c0593805c6d6" dependencies = [ "futures-core", - "futures-io", "futures-macro", "futures-sink", "futures-task", diff --git a/src/components/file_explorer.rs b/src/components/file_explorer.rs index b8078f7..4de4128 100644 --- a/src/components/file_explorer.rs +++ b/src/components/file_explorer.rs @@ -403,7 +403,7 @@ mod tests { explorer .files .iter() - .all(|p| !p.extension().is_some_and(|e| e == "txt")), + .all(|p| p.extension().is_none_or(|e| e != "txt")), "txt files must be hidden in dir-select mode" ); // Subdirectories must still appear. diff --git a/src/components/p2pool_client.rs b/src/components/p2pool_client.rs index d063760..e548f79 100644 --- a/src/components/p2pool_client.rs +++ b/src/components/p2pool_client.rs @@ -158,12 +158,12 @@ impl P2PoolClient { { Ok(data) => Ok(data), Err(error) => { - if self.should_try_fallback(&error) { - if let Some(fallback_base_url) = &self.fallback_base_url { - return self - .fetch_json_from_base_url(fallback_base_url, path, query, false) - .await; - } + if self.should_try_fallback(&error) + && let Some(fallback_base_url) = &self.fallback_base_url + { + return self + .fetch_json_from_base_url(fallback_base_url, path, query, false) + .await; } Err(error) } @@ -187,10 +187,8 @@ impl P2PoolClient { request = request.query(query); } - if use_auth { - if let Some((user, pass)) = &self.auth_credentials { - request = request.basic_auth(user, Some(pass)); - } + if use_auth && let Some((user, pass)) = &self.auth_credentials { + request = request.basic_auth(user, Some(pass)); } let response = request.send().await?.error_for_status()?; diff --git a/src/components/p2pool_config_view.rs b/src/components/p2pool_config_view.rs index f7c0f9b..2a5e4b0 100644 --- a/src/components/p2pool_config_view.rs +++ b/src/components/p2pool_config_view.rs @@ -153,7 +153,7 @@ impl P2PoolConfigView { let entries: Vec = app .p2pool_config .as_ref() - .map(|cfg| flatten_config(cfg)) + .map(flatten_config) .unwrap_or_default(); // Status bar (warning or save message) @@ -404,10 +404,7 @@ port = 3030 .unwrap(); // keep dir alive until Config is loaded - let cfg = Config::load(path.to_str().unwrap()).expect("inline test config must parse"); - - // dir drops here but we already have cfg - cfg + Config::load(path.to_str().unwrap()).expect("inline test config must parse") } #[test] @@ -586,11 +583,16 @@ port = 3030 fn render_no_path_shows_prompt() { let backend = TestBackend::new(80, 10); let mut terminal = Terminal::new(backend).unwrap(); - let mut app = App::default(); - app.p2pool_conf_path = None; - app.p2pool_config_view.warning_message = None; + let mut app = App { + p2pool_conf_path: None, + p2pool_config_view: P2PoolConfigView { + warning_message: None, + ..Default::default() + }, + ..Default::default() + }; terminal - .draw(|f| P2PoolConfigView::render(f, &mut app, f.size())) + .draw(|f| P2PoolConfigView::render(f, &mut app, f.area())) .unwrap(); assert!(buffer_text(&terminal).contains("Press [Enter] to select")); } @@ -599,11 +601,16 @@ port = 3030 fn render_no_path_shows_warning_message() { let backend = TestBackend::new(80, 10); let mut terminal = Terminal::new(backend).unwrap(); - let mut app = App::default(); - app.p2pool_conf_path = None; - app.p2pool_config_view.warning_message = Some("File not found".into()); + let mut app = App { + p2pool_conf_path: None, + p2pool_config_view: P2PoolConfigView { + warning_message: Some("File not found".into()), + ..Default::default() + }, + ..Default::default() + }; terminal - .draw(|f| P2PoolConfigView::render(f, &mut app, f.size())) + .draw(|f| P2PoolConfigView::render(f, &mut app, f.area())) .unwrap(); assert!(buffer_text(&terminal).contains("File not found")); } @@ -616,9 +623,10 @@ port = 3030 let backend = TestBackend::new(120, 30); let mut terminal = Terminal::new(backend).unwrap(); - let mut app = App::default(); - - app.p2pool_conf_path = Some(PathBuf::from("test.toml")); + let mut app = App { + p2pool_conf_path: Some(PathBuf::from("test.toml")), + ..Default::default() + }; let cfg = make_config(); app.p2pool_config = Some(cfg.clone()); @@ -633,7 +641,7 @@ port = 3030 app.p2pool_config_view.edit_input = "secret123".into(); terminal - .draw(|f| P2PoolConfigView::render(f, &mut app, f.size())) + .draw(|f| P2PoolConfigView::render(f, &mut app, f.area())) .unwrap(); let text = buffer_text(&terminal); diff --git a/src/components/p2pool_status_view.rs b/src/components/p2pool_status_view.rs index 9419de0..157eb01 100644 --- a/src/components/p2pool_status_view.rs +++ b/src/components/p2pool_status_view.rs @@ -332,7 +332,7 @@ impl P2PoolStatusView { return "-".to_string(); } - let difficulty = (0x00ff_ff_u32 as f64 / mantissa as f64) * 256_f64.powi(0x1d - exponent); + let difficulty = (0x0000_ffff_u32 as f64 / mantissa as f64) * 256_f64.powi(0x1d - exponent); if !difficulty.is_finite() || difficulty <= 0.0 { return "-".to_string(); } @@ -430,9 +430,15 @@ impl Default for P2PoolStatusView { #[cfg(test)] mod tests { use super::*; - use crate::components::p2pool_client::{ChainInfo, PeerInfo}; + use crate::components::p2pool_client::{ + ChainInfo, PeerInfo, ShareInfo, SharesResponse, UncleInfo, + }; + use crate::components::p2pool_websocket::{LivePeerEvent, LiveShare}; use ratatui::{Terminal, backend::TestBackend, prelude::Rect}; + const SHARE_TAB: usize = 1; + const PEER_TAB: usize = 2; + fn render_view(app: &App) -> String { let backend = TestBackend::new(100, 25); let mut terminal = Terminal::new(backend).unwrap(); @@ -451,6 +457,61 @@ mod tests { .collect() } + fn share_info( + height: u64, + blockhash: &str, + miner_address: &str, + timestamp: u64, + bits: &str, + uncle_count: usize, + ) -> ShareInfo { + ShareInfo { + blockhash: blockhash.to_string(), + prev_blockhash: format!("{blockhash}-prev"), + height, + miner_address: miner_address.to_string(), + timestamp, + bits: bits.to_string(), + uncles: (0..uncle_count) + .map(|index| UncleInfo { + blockhash: format!("{blockhash}-uncle-{index}"), + prev_blockhash: format!("{blockhash}-uncle-prev-{index}"), + miner_address: miner_address.to_string(), + timestamp, + height, + }) + .collect(), + } + } + + fn live_share( + height: u64, + blockhash: &str, + miner_address: &str, + timestamp: u64, + bits: &str, + uncle_count: usize, + ) -> LiveShare { + LiveShare { + blockhash: blockhash.to_string(), + prev_blockhash: format!("{blockhash}-prev"), + height, + miner_address: miner_address.to_string(), + timestamp, + bits: bits.to_string(), + uncles: (0..uncle_count) + .map(|index| format!("{blockhash}-uncle-{index}")) + .collect(), + } + } + + #[test] + fn default_constructs_status_view() { + let view = P2PoolStatusView; + + assert_eq!(format!("{view:?}"), "P2PoolStatusView"); + } + #[test] fn render_dispatches_chain_info_for_tab_zero() { let app = App::new(); @@ -461,9 +522,78 @@ mod tests { } #[test] - fn render_dispatches_peer_info_for_tab_one() { + fn render_dispatches_share_info_for_tab_one() { + let mut app = App::new(); + app.p2pool_status_tab = SHARE_TAB; + + let output = render_view(&app); + + assert!(output.contains("Difficulty")); + assert!(output.contains("Uncles")); + } + + #[test] + fn render_share_info_lists_api_and_live_shares_without_duplicates() { + let mut app = App::new(); + app.p2pool_status_tab = SHARE_TAB; + app.live_shares = vec![live_share( + 42, + "livehash", + "olderlive", + 1_700_000_000, + "1d00ffff", + 1, + )]; + app.share_info = Some(SharesResponse { + from_height: 40, + to_height: 42, + shares: vec![ + share_info(42, "apihash", "newerapi", 1_700_000_060, "1e00ffff", 2), + share_info(41, "livehash", "dupeapi", 1_700_000_120, "1d00ffff", 0), + ], + }); + + let output = render_view(&app); + + assert!(output.contains("newerapi")); + assert!(output.contains("olderlive")); + assert!(!output.contains("dupeapi")); + assert!(output.contains("0.0039")); + assert!(output.contains("1.00")); + assert!(output.contains("11/14/2023, 10:14:20 PM")); + assert!(output.find("newerapi").unwrap() < output.find("olderlive").unwrap()); + } + + #[test] + fn render_share_info_shows_recent_share_error_before_live_error() { let mut app = App::new(); - app.p2pool_status_tab = 1; + app.p2pool_status_tab = SHARE_TAB; + app.p2pool_share_info_error = Some("recent fetch failed".to_string()); + app.p2pool_live_error = Some("websocket closed".to_string()); + + let output = render_view(&app); + + assert!(output.contains("Recent")); + assert!(!output.contains("Live sh")); + assert!(!output.contains("Waiting")); + } + + #[test] + fn render_share_info_shows_live_error_when_recent_fetch_has_not_failed() { + let mut app = App::new(); + app.p2pool_status_tab = SHARE_TAB; + app.p2pool_live_error = Some("websocket closed".to_string()); + + let output = render_view(&app); + + assert!(output.contains("Live sh")); + assert!(!output.contains("Waiting")); + } + + #[test] + fn render_dispatches_peer_info_for_tab_two() { + let mut app = App::new(); + app.p2pool_status_tab = PEER_TAB; let output = render_view(&app); @@ -535,7 +665,7 @@ mod tests { #[test] fn render_peer_info_shows_loading_state_with_no_data() { let mut app = App::new(); - app.p2pool_status_tab = 1; + app.p2pool_status_tab = PEER_TAB; let output = render_view(&app); @@ -545,7 +675,7 @@ mod tests { #[test] fn render_peer_info_shows_error_when_fetch_failed() { let mut app = App::new(); - app.p2pool_status_tab = 1; + app.p2pool_status_tab = PEER_TAB; app.p2pool_peer_info_error = Some("request timed out".to_string()); let output = render_view(&app); @@ -557,7 +687,7 @@ mod tests { #[test] fn render_peer_info_shows_empty_state_when_no_peers_are_connected() { let mut app = App::new(); - app.p2pool_status_tab = 1; + app.p2pool_status_tab = PEER_TAB; app.peer_info = Some(Vec::new()); let output = render_view(&app); @@ -568,7 +698,7 @@ mod tests { #[test] fn render_peer_info_lists_connected_peers_with_statuses() { let mut app = App::new(); - app.p2pool_status_tab = 1; + app.p2pool_status_tab = PEER_TAB; app.peer_info = Some(vec![ PeerInfo { peer_id: "12D3KooWPeerOne".to_string(), @@ -587,10 +717,34 @@ mod tests { assert!(output.contains("12D3KooWPeerTwo (Syncing)")); } + #[test] + fn render_peer_info_appends_live_error_and_recent_peer_events() { + let mut app = App::new(); + app.p2pool_status_tab = PEER_TAB; + app.peer_info = Some(Vec::new()); + app.p2pool_live_error = Some("websocket closed".to_string()); + app.live_peer_events = (1..=9) + .map(|index| LivePeerEvent { + peer_id: format!("12D3KooWPeerEvent{index}"), + status: format!("Status{index}"), + }) + .collect(); + + let output = render_view(&app); + + assert!(output.contains("No connected peers")); + assert!(output.contains("Live stream error: websocket closed")); + assert!(output.contains("Live Peer Events")); + assert!(output.contains("Status9: 12D3KooWPeerEvent9")); + assert!(output.contains("Status2: 12D3KooWPeerEvent2")); + assert!(!output.contains("Status1: 12D3KooWPeerEvent1")); + assert!(output.find("Status9").unwrap() < output.find("Status2").unwrap()); + } + #[test] fn render_peer_info_defaults_missing_status_to_connected() { let mut app = App::new(); - app.p2pool_status_tab = 1; + app.p2pool_status_tab = PEER_TAB; app.peer_info = Some(vec![PeerInfo { peer_id: "12D3KooWNoStatus".to_string(), status: None, @@ -601,4 +755,54 @@ mod tests { assert!(output.contains("Connected Peers : 1")); assert!(output.contains("12D3KooWNoStatus (Connected)")); } + + #[test] + fn short_value_preserves_short_values_and_truncates_long_values() { + assert_eq!(P2PoolStatusView::short_value("short", 10), "short"); + assert_eq!( + P2PoolStatusView::short_value("abcdefghijklmnop", 10), + "abc...mnop" + ); + assert_eq!(P2PoolStatusView::short_value("abcdef", 3), "abc"); + } + + #[test] + fn format_difficulty_formats_valid_bits_values() { + assert_eq!(P2PoolStatusView::format_difficulty("1d00ffff"), "1.00"); + assert_eq!(P2PoolStatusView::format_difficulty("486604799"), "1.00"); + assert_eq!(P2PoolStatusView::format_difficulty("0X1D00FFFF"), "1.00"); + assert_eq!(P2PoolStatusView::format_difficulty("1b00ffff"), "65,536"); + assert_eq!(P2PoolStatusView::format_difficulty("1e00ffff"), "0.0039"); + } + + #[test] + fn format_difficulty_handles_invalid_or_unusable_bits() { + assert_eq!(P2PoolStatusView::format_difficulty(""), ""); + assert_eq!( + P2PoolStatusView::format_difficulty("not-a-compact-bits-value"), + "not...alue" + ); + assert_eq!(P2PoolStatusView::format_difficulty("1d000000"), "-"); + assert_eq!(P2PoolStatusView::format_difficulty("ff00ffff"), "-"); + } + + #[test] + fn format_timestamp_formats_seconds_and_milliseconds() { + assert_eq!( + P2PoolStatusView::format_timestamp(0), + "1/1/1970, 12:00:00 AM" + ); + assert_eq!( + P2PoolStatusView::format_timestamp(43_200), + "1/1/1970, 12:00:00 PM" + ); + assert_eq!( + P2PoolStatusView::format_timestamp(1_700_000_000), + "11/14/2023, 10:13:20 PM" + ); + assert_eq!( + P2PoolStatusView::format_timestamp(1_700_000_000_000), + "11/14/2023, 10:13:20 PM" + ); + } } diff --git a/src/components/p2pool_websocket.rs b/src/components/p2pool_websocket.rs index fa3f967..23d0318 100644 --- a/src/components/p2pool_websocket.rs +++ b/src/components/p2pool_websocket.rs @@ -130,34 +130,58 @@ impl P2PoolWebSocketClient { fn ws_url_with_auth(&self, path: &str) -> Result { let mut url = self.ws_url(path)?; + self.apply_auth(&mut url); + Ok(url) + } + + fn ws_urls_with_auth(&self, path: &str) -> Result> { + let mut urls = vec![self.ws_url_with_auth(path)?]; + if urls[0].scheme() == "ws" { + let mut wss_url = self.ws_url(path)?; + self.apply_auth(&mut wss_url); + wss_url.set_scheme("wss").unwrap(); + urls.push(wss_url); + } + Ok(urls) + } + + fn apply_auth(&self, url: &mut Url) { if let Some((user, pass)) = &self.auth_credentials { let token = STANDARD.encode(format!("{}:{}", user, pass)); url.query_pairs_mut().append_pair("token", &token); } - Ok(url) } pub async fn subscribe_live_events( &self, tx: mpsc::UnboundedSender>, ) -> anyhow::Result<()> { - let url = self.ws_url_with_auth("/ws")?; - match self.subscribe_live_events_at(url, tx.clone()).await { - Ok(()) => Ok(()), - Err(primary_error) => { - if let Some(fallback_base_url) = &self.fallback_base_url { - let fallback_url = self.ws_url_from_base_url(fallback_base_url, "/ws")?; - if self - .subscribe_live_events_at(fallback_url, tx) - .await - .is_ok() - { - return Ok(()); + let urls = self.ws_urls_with_auth("/ws")?; + let mut primary_error = None; + + for url in urls { + match self.subscribe_live_events_at(url, tx.clone()).await { + Ok(()) => return Ok(()), + Err(error) => { + if primary_error.is_none() { + primary_error = Some(error); } } - Err(primary_error) } } + + if let Some(fallback_base_url) = &self.fallback_base_url { + let fallback_url = self.ws_url_from_base_url(fallback_base_url, "/ws")?; + if self + .subscribe_live_events_at(fallback_url, tx) + .await + .is_ok() + { + return Ok(()); + } + } + + Err(primary_error.unwrap_or_else(|| anyhow::anyhow!("websocket connection failed"))) } async fn subscribe_live_events_at( @@ -237,6 +261,11 @@ where #[cfg(test)] mod tests { use super::*; + use futures_util::StreamExt; + use tokio::net::TcpListener; + use tokio::sync::mpsc; + use tokio_tungstenite::accept_async; + use tokio_tungstenite::tungstenite::Message; #[test] fn ws_url_converts_http_to_ws_and_encodes_auth_token() { @@ -314,4 +343,159 @@ mod tests { assert!(matches!(event, WebSocketEvent::Peer(_))); } + + #[test] + fn ws_urls_with_auth_try_plain_and_secure_websocket_schemes() { + let client = P2PoolWebSocketClient::with_base_url("http://127.0.0.1:46884") + .with_auth("user".into(), "password".into()); + + let urls = client.ws_urls_with_auth("/ws").unwrap(); + + assert_eq!(urls.len(), 2); + assert_eq!( + urls[0].as_str(), + "ws://127.0.0.1:46884/ws?token=dXNlcjpwYXNzd29yZA%3D%3D" + ); + assert_eq!( + urls[1].as_str(), + "wss://127.0.0.1:46884/ws?token=dXNlcjpwYXNzd29yZA%3D%3D" + ); + } + + #[tokio::test] + async fn subscribe_live_events_emits_share_and_peer_events() { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + + let server = tokio::spawn(async move { + let (stream, _) = listener.accept().await.unwrap(); + let websocket = accept_async(stream).await.unwrap(); + let (mut write, mut read) = websocket.split(); + + for _ in 0..2 { + let _ = read.next().await.unwrap().unwrap(); + } + + write.send(Message::Binary(vec![1, 2, 3])).await.unwrap(); + write + .send(Message::Text( + serde_json::json!({ + "topic": "Share", + "data": { + "blockhash": "0000", + "prev_blockhash": "ffff", + "height": 42, + "miner_address": "miner", + "timestamp": 1700000000, + "bits": "1d00ffff", + "uncles": ["aaaa"] + } + }) + .to_string(), + )) + .await + .unwrap(); + write + .send(Message::Text( + serde_json::json!({ + "topic": "Peer", + "data": { + "peer_id": "12D3KooWPeerOne", + "status": "Connected" + } + }) + .to_string(), + )) + .await + .unwrap(); + let _ = write.close().await; + }); + + let client = P2PoolWebSocketClient::with_base_url(format!("http://{addr}")); + let (tx, mut rx) = mpsc::unbounded_channel(); + let subscribe_handle = tokio::spawn(async move { client.subscribe_live_events(tx).await }); + + let first_event = rx.recv().await.unwrap().unwrap(); + let second_event = rx.recv().await.unwrap().unwrap(); + let result = subscribe_handle.await.unwrap(); + server.await.unwrap(); + + assert!(result.is_ok()); + assert!(matches!(first_event, LiveP2PoolEvent::Share(_))); + assert!(matches!(second_event, LiveP2PoolEvent::Peer(_))); + } + + #[tokio::test] + async fn subscribe_live_events_emits_error_for_invalid_json() { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + + let server = tokio::spawn(async move { + let (stream, _) = listener.accept().await.unwrap(); + let websocket = accept_async(stream).await.unwrap(); + let (mut write, mut read) = websocket.split(); + + let _ = read.next().await.unwrap().unwrap(); + let _ = read.next().await.unwrap().unwrap(); + write.send(Message::Text("not-json".into())).await.unwrap(); + let _ = write.close().await; + }); + + let client = P2PoolWebSocketClient::with_base_url(format!("http://{addr}")); + let (tx, mut rx) = mpsc::unbounded_channel(); + let subscribe_handle = tokio::spawn(async move { client.subscribe_live_events(tx).await }); + + let event = rx.recv().await.unwrap(); + let result = subscribe_handle.await.unwrap(); + server.await.unwrap(); + + assert!(result.is_ok()); + assert!(event.is_err()); + } + + #[tokio::test] + async fn subscribe_live_events_uses_fallback_base_url_on_connection_failure() { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let fallback_addr = listener.local_addr().unwrap(); + + let server = tokio::spawn(async move { + let (stream, _) = listener.accept().await.unwrap(); + let websocket = accept_async(stream).await.unwrap(); + let (mut write, mut read) = websocket.split(); + + let _ = read.next().await.unwrap().unwrap(); + let _ = read.next().await.unwrap().unwrap(); + write + .send(Message::Text( + serde_json::json!({ + "topic": "Share", + "data": { + "blockhash": "fallback", + "prev_blockhash": "prev", + "height": 7, + "miner_address": "miner", + "timestamp": 1700000000, + "bits": "1d00ffff", + "uncles": [] + } + }) + .to_string(), + )) + .await + .unwrap(); + let _ = write.close().await; + }); + + let client = P2PoolWebSocketClient::with_base_url("http://127.0.0.1:1") + .with_fallback_base_url(format!("http://{fallback_addr}")); + let (tx, mut rx) = mpsc::unbounded_channel(); + let subscribe_handle = tokio::spawn(async move { client.subscribe_live_events(tx).await }); + + let event = rx.recv().await.unwrap().unwrap(); + let result = subscribe_handle.await.unwrap(); + server.await.unwrap(); + + assert!(result.is_ok()); + assert!(matches!(event, LiveP2PoolEvent::Share(_))); + } } diff --git a/src/config.rs b/src/config.rs index a70f086..3e74870 100644 --- a/src/config.rs +++ b/src/config.rs @@ -6,7 +6,7 @@ use anyhow::Result; use config::{Config, File}; const DEFAULT_API_HOST: &str = "127.0.0.1"; -const DEFAULT_API_PORT: u16 = 9332; +const DEFAULT_API_PORT: u16 = 46884; #[derive(Debug, Clone)] pub struct ApiConfig { diff --git a/src/main.rs b/src/main.rs index 4053932..5b0acca 100644 --- a/src/main.rs +++ b/src/main.rs @@ -24,8 +24,6 @@ use crossterm::{ }; use ratatui::{Terminal, backend::Backend, backend::CrosstermBackend}; use std::{io, time::Duration}; -use std::io; -use std::time::Duration; #[tokio::main] async fn main() -> Result<()> { @@ -89,15 +87,15 @@ where app.poll_live_p2pool_events(); terminal.draw(|f| ui::ui(f, app))?; - if event::poll(Duration::from_millis(100))? { - if let Event::Key(key) = event::read()? { - match dispatch_key(key, app) { - KeyOutcome::Ignored => {} - KeyOutcome::Exit => return Ok(()), - KeyOutcome::Action(action) => { - if handle_action(action, app)?.is_break() { - return Ok(()); - } + if event::poll(Duration::from_millis(100))? + && let Event::Key(key) = event::read()? + { + match dispatch_key(key, app) { + KeyOutcome::Ignored => {} + KeyOutcome::Exit => return Ok(()), + KeyOutcome::Action(action) => { + if handle_action(action, app)?.is_break() { + return Ok(()); } } } @@ -201,7 +199,7 @@ fn dispatch_key(key: event::KeyEvent, app: &mut App) -> KeyOutcome { let entries = app .p2pool_config .as_ref() - .map(|cfg| flatten_config(cfg)) + .map(flatten_config) .unwrap_or_default(); app.p2pool_config_view.handle_input(key, &entries) } @@ -250,17 +248,17 @@ fn bootstrap_from_settings(app: &mut App) { } // P2Pool config — only set the path when the config is actually loadable - if let Some(path) = &app.settings.p2pool_conf_path.clone() { - if let Some(p) = path.to_str() { - match P2PoolConfig::load(p) { - Ok(cfg) => { - app.p2pool_conf_path = Some(path.clone()); - app.p2pool_config = Some(cfg); - } - Err(e) => { - eprintln!("pdm: failed to load p2pool config on startup: {e}"); - // Leave both as None so the view prompts the user to re-select - } + if let Some(path) = &app.settings.p2pool_conf_path.clone() + && let Some(p) = path.to_str() + { + match P2PoolConfig::load(p) { + Ok(cfg) => { + app.p2pool_conf_path = Some(path.clone()); + app.p2pool_config = Some(cfg); + } + Err(e) => { + eprintln!("pdm: failed to load p2pool config on startup: {e}"); + // Leave both as None so the view prompts the user to re-select } } } @@ -286,7 +284,7 @@ fn handle_action(action: AppAction, app: &mut App) -> Result> { AppAction::OpenExplorerForSettings(field) => { let dir_select = FIELDS .get(field) - .map_or(false, |f| matches!(f.1, FieldKind::DirectoryPicker)); + .is_some_and(|f| matches!(f.1, FieldKind::DirectoryPicker)); if app.explorer.allow_dir_select != dir_select { app.explorer.allow_dir_select = dir_select; app.explorer.load_directory(); @@ -1978,11 +1976,7 @@ port = 46884 #[test] fn commit_p2pool_edit_success_clears_warning() { - use std::path::PathBuf; - let mut app = App::new(); - let path = PathBuf::from("dummy.toml"); - write_valid_p2pool_toml(&path); let dir = tempfile::tempdir().unwrap(); let file = dir.path().join("p2pool.toml"); diff --git a/src/p2poolv2_config.rs b/src/p2poolv2_config.rs index 141ea5b..300e04c 100644 --- a/src/p2poolv2_config.rs +++ b/src/p2poolv2_config.rs @@ -710,10 +710,7 @@ port = 3030 .unwrap(); // keep dir alive until Config is loaded - let cfg = Config::load(path.to_str().unwrap()).expect("inline test config must parse"); - - // dir drops here but we already have cfg - cfg + Config::load(path.to_str().unwrap()).expect("inline test config must parse") } #[test] From 40c848ac51364cad945681c6d43bf55a85f3425f Mon Sep 17 00:00:00 2001 From: Raunak Kumar Date: Sat, 4 Jul 2026 16:56:01 +0000 Subject: [PATCH 4/4] fix: preserve P2Pool fallback auth and websocket handling --- src/components/p2pool_client.rs | 24 ++++++- src/components/p2pool_websocket.rs | 104 +++++++++++++++++++++-------- 2 files changed, 101 insertions(+), 27 deletions(-) diff --git a/src/components/p2pool_client.rs b/src/components/p2pool_client.rs index e548f79..a7741db 100644 --- a/src/components/p2pool_client.rs +++ b/src/components/p2pool_client.rs @@ -162,7 +162,7 @@ impl P2PoolClient { && let Some(fallback_base_url) = &self.fallback_base_url { return self - .fetch_json_from_base_url(fallback_base_url, path, query, false) + .fetch_json_from_base_url(fallback_base_url, path, query, true) .await; } Err(error) @@ -258,6 +258,28 @@ mod tests { assert_eq!(client.fallback_base_url.as_deref(), Some(FALLBACK_BASE_URL)); } + #[tokio::test] + async fn fallback_fetch_uses_basic_auth_when_configured() { + let mut server = Server::new_async().await; + + let mock = server + .mock("GET", "/chain_info") + .match_header("authorization", "Basic dXNlcjpwYXNzd29yZA==") + .with_status(200) + .with_header("content-type", "application/json") + .with_body(json!({ "total_work": "abc" }).to_string()) + .create(); + + let client = P2PoolClient::with_base_url("http://127.0.0.1:1") + .with_fallback_base_url(server.url()) + .with_auth("user".into(), "password".into()); + + let result = client.fetch_chain_info().await.unwrap(); + + assert_eq!(result.total_work, "abc"); + mock.assert(); + } + #[tokio::test] async fn test_fetch_chain_info_success() { let mut server = Server::new_async().await; diff --git a/src/components/p2pool_websocket.rs b/src/components/p2pool_websocket.rs index 23d0318..b0e70fb 100644 --- a/src/components/p2pool_websocket.rs +++ b/src/components/p2pool_websocket.rs @@ -124,7 +124,14 @@ impl P2PoolWebSocketClient { _ => {} } - url.set_path(path); + let base_path = url.path().trim_end_matches('/'); + let extra_path = path.trim_start_matches('/'); + let full_path = if base_path.is_empty() || base_path == "/" { + format!("/{extra_path}") + } else { + format!("{base_path}/{extra_path}") + }; + url.set_path(&full_path); Ok(url) } @@ -171,7 +178,8 @@ impl P2PoolWebSocketClient { } if let Some(fallback_base_url) = &self.fallback_base_url { - let fallback_url = self.ws_url_from_base_url(fallback_base_url, "/ws")?; + let mut fallback_url = self.ws_url_from_base_url(fallback_base_url, "/ws")?; + self.apply_auth(&mut fallback_url); if self .subscribe_live_events_at(fallback_url, tx) .await @@ -202,35 +210,43 @@ impl P2PoolWebSocketClient { } while let Some(message_result) = read.next().await { - let message = message_result?; - if let Message::Text(text) = message { - match serde_json::from_str::(&text) { - Ok(WebSocketEvent::Share(data)) => { - let live_share = LiveShare { - blockhash: data.blockhash, - prev_blockhash: data.prev_blockhash, - height: data.height, - miner_address: data.miner_address, - timestamp: data.timestamp, - bits: data.bits, - uncles: data.uncles, - }; - let _ = tx.send(Ok(LiveP2PoolEvent::Share(live_share))); - } - Ok(WebSocketEvent::Peer(data)) => { - let live_peer = LivePeerEvent { - peer_id: data.peer_id, - status: data.status, - }; - let _ = tx.send(Ok(LiveP2PoolEvent::Peer(live_peer))); - } - Err(error) => { - let _ = tx.send(Err(anyhow::Error::new(error))); + match message_result { + Ok(message) => { + if let Message::Text(text) = message { + match serde_json::from_str::(&text) { + Ok(WebSocketEvent::Share(data)) => { + let live_share = LiveShare { + blockhash: data.blockhash, + prev_blockhash: data.prev_blockhash, + height: data.height, + miner_address: data.miner_address, + timestamp: data.timestamp, + bits: data.bits, + uncles: data.uncles, + }; + let _ = tx.send(Ok(LiveP2PoolEvent::Share(live_share))); + } + Ok(WebSocketEvent::Peer(data)) => { + let live_peer = LivePeerEvent { + peer_id: data.peer_id, + status: data.status, + }; + let _ = tx.send(Ok(LiveP2PoolEvent::Peer(live_peer))); + } + Err(error) => { + let _ = tx.send(Err(anyhow::Error::new(error))); + } + } } } + Err(error) => { + let _ = tx.send(Err(anyhow::Error::new(error))); + return Ok(()); + } } } + let _ = tx.send(Err(anyhow::anyhow!("websocket connection closed"))); Ok(()) } } @@ -289,6 +305,15 @@ mod tests { assert_eq!(url.as_str(), "wss://127.0.0.1:46884/ws"); } + #[test] + fn ws_url_appends_path_to_existing_base_path() { + let client = P2PoolWebSocketClient::with_base_url("http://127.0.0.1:46884/api"); + + let url = client.ws_url("/ws").unwrap(); + + assert_eq!(url.as_str(), "ws://127.0.0.1:46884/api/ws"); + } + #[test] fn websocket_event_accepts_share_messages() { let event: WebSocketEvent = serde_json::from_value(serde_json::json!({ @@ -425,6 +450,33 @@ mod tests { assert!(matches!(second_event, LiveP2PoolEvent::Peer(_))); } + #[tokio::test] + async fn subscribe_live_events_emits_error_when_connection_closes() { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + + let server = tokio::spawn(async move { + let (stream, _) = listener.accept().await.unwrap(); + let websocket = accept_async(stream).await.unwrap(); + let (mut write, mut read) = websocket.split(); + + let _ = read.next().await.unwrap().unwrap(); + let _ = read.next().await.unwrap().unwrap(); + let _ = write.close().await; + }); + + let client = P2PoolWebSocketClient::with_base_url(format!("http://{addr}")); + let (tx, mut rx) = mpsc::unbounded_channel(); + let subscribe_handle = tokio::spawn(async move { client.subscribe_live_events(tx).await }); + + let event = rx.recv().await.unwrap(); + let result = subscribe_handle.await.unwrap(); + server.await.unwrap(); + + assert!(event.is_err()); + assert!(result.is_ok()); + } + #[tokio::test] async fn subscribe_live_events_emits_error_for_invalid_json() { let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();