diff --git a/crates/wallet-app/src/commands.rs b/crates/wallet-app/src/commands.rs index e401da3..b60cf7f 100644 --- a/crates/wallet-app/src/commands.rs +++ b/crates/wallet-app/src/commands.rs @@ -15,10 +15,11 @@ use wallet_core::ss58; use wallet_core::{Keystore, KeystoreError}; use wallet_data::keys::seed_from_mnemonic; use wallet_data::mnemonic; +use wallet_data::substrate::ChainManager; use wallet_entities::{ AppInfo, CreationChallenge, CreationStart, SignatureScheme, WalletError, WalletSummary, }; -use wallet_entities::{ChainProfile, SessionAccountInfo, SessionStatusInfo}; +use wallet_entities::{ChainId, ChainProfile, ChainStatus, SessionAccountInfo, SessionStatusInfo}; use crate::state::{AppState, PendingCreation}; @@ -223,3 +224,53 @@ pub fn session_status(state: State<'_, AppState>) -> Result) -> Result, WalletError> { Ok(state.profiles.iter().cloned().collect()) } + +/// Start supervising a chain's connection, if not already. Idempotent; the +/// status command says how it is going. +#[tauri::command] +pub async fn chain_connect( + state: State<'_, AppState>, + chain: ChainId, +) -> Result { + let profile = state + .profiles + .get(&chain) + .cloned() + .ok_or_else(|| WalletError::NotFound(format!("chain {}", chain.0)))?; + let manager = { + let mut chains = state + .chains + .lock() + .map_err(|_| WalletError::Internal("state poisoned".into()))?; + chains + .entry(chain) + .or_insert_with(|| ChainManager::start(profile)) + .clone() + }; + Ok(manager.status()) +} + +#[tauri::command] +pub fn chain_status( + state: State<'_, AppState>, + chain: ChainId, +) -> Result { + let chains = state + .chains + .lock() + .map_err(|_| WalletError::Internal("state poisoned".into()))?; + match chains.get(&chain) { + Some(m) => Ok(m.status()), + None => Ok(ChainStatus { + chain, + connected: false, + endpoint: None, + spec_version: None, + transaction_version: None, + finalized_number: None, + head_age_seconds: None, + last_error: None, + failures: 0, + }), + } +} diff --git a/crates/wallet-app/src/lib.rs b/crates/wallet-app/src/lib.rs index 8f2e78c..4bea655 100644 --- a/crates/wallet-app/src/lib.rs +++ b/crates/wallet-app/src/lib.rs @@ -48,6 +48,8 @@ pub fn run() { commands::lock, commands::session_status, commands::chains_list, + commands::chain_connect, + commands::chain_status, ]) .run(tauri::generate_context!()) .expect("the Tauri runtime failed to start"); diff --git a/crates/wallet-app/src/state.rs b/crates/wallet-app/src/state.rs index c976bfe..a3622d5 100644 --- a/crates/wallet-app/src/state.rs +++ b/crates/wallet-app/src/state.rs @@ -3,6 +3,7 @@ //! cancelled, and later the unlock session. Everything is behind a mutex //! because Tauri commands run concurrently. +use std::collections::HashMap; use std::sync::Mutex; use std::time::Duration; @@ -10,6 +11,8 @@ use wallet_core::backup::BackupChallenge; use wallet_core::profiles::ProfileRegistry; use wallet_core::session::Session; use wallet_data::keystore::FileKeystore; +use wallet_data::substrate::ChainManager; +use wallet_entities::ChainId; use wallet_entities::SignatureScheme; use zeroize::Zeroizing; @@ -29,6 +32,10 @@ pub struct AppState { pub pending_creation: Mutex>, pub session: Session, pub profiles: ProfileRegistry, + /// One supervisor per chain the UI has asked to connect to. Started on + /// demand, never at boot: a profile whose endpoint does not exist yet + /// would otherwise retry forever for nobody. + pub chains: Mutex>, } impl AppState { @@ -63,6 +70,7 @@ impl AppState { pending_creation: Mutex::new(None), session: Session::new(DEFAULT_IDLE_LOCK), profiles, + chains: Mutex::new(HashMap::new()), }) } } diff --git a/crates/wallet-data/Cargo.toml b/crates/wallet-data/Cargo.toml index 078d287..ca7be81 100644 --- a/crates/wallet-data/Cargo.toml +++ b/crates/wallet-data/Cargo.toml @@ -31,6 +31,7 @@ qp-human-checkphrase.workspace = true subxt.workspace = true parity-scale-codec.workspace = true hex.workspace = true +tokio.workspace = true [dev-dependencies] tokio.workspace = true diff --git a/crates/wallet-data/src/substrate/connection.rs b/crates/wallet-data/src/substrate/connection.rs new file mode 100644 index 0000000..b732432 --- /dev/null +++ b/crates/wallet-data/src/substrate/connection.rs @@ -0,0 +1,381 @@ +//! The connection to one chain: which of the profile's endpoints is in use, +//! what the node last said, and a supervisor that keeps it that way. +//! +//! The rules: endpoints are tried in the profile's order; a node whose +//! genesis is not the profile's is refused, not merely skipped, so a typo in +//! an override file cannot quietly sign for the wrong chain; after any +//! failure the next endpoint is tried after a jittered backoff; the runtime +//! version is read at connect and refreshed on every runtime upgrade, so the +//! signing context and the metadata can never be stale; finalized heads are +//! followed so the status can say how old the last head is. + +use std::sync::{Arc, RwLock}; +use std::time::{Duration, Instant}; + +use subxt::OnlineClient; +use subxt::backend::legacy::LegacyRpcMethods; +use subxt::backend::rpc::RpcClient; +use wallet_core::ChainError; +use wallet_core::profiles::genesis_matches; +use wallet_entities::{ChainProfile, ChainStatus}; + +use super::config::QuantusConfig; +use super::offline::ChainState; + +/// Exponential backoff with full jitter, capped. Pure, so the schedule is +/// testable without sleeping. `next` takes the jitter as a fraction so the +/// caller supplies the randomness. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct Backoff { + pub base: Duration, + pub cap: Duration, + pub failures: u32, +} + +impl Backoff { + pub fn new(base: Duration, cap: Duration) -> Self { + Self { + base, + cap, + failures: 0, + } + } + + /// The delay before the next attempt, given `jitter` in `0.0..=1.0`. + /// Doubles per failure up to the cap, then multiplies by `0.5 + jitter/2` + /// so retries from many wallets do not land on a node together. + pub fn delay(&self, jitter: f64) -> Duration { + let exp = self.base.saturating_mul(1u32 << self.failures.min(16)); + let capped = exp.min(self.cap); + capped.mul_f64(0.5 + jitter.clamp(0.0, 1.0) / 2.0) + } + + pub fn fail(&mut self) { + self.failures = self.failures.saturating_add(1); + } + + pub fn reset(&mut self) { + self.failures = 0; + } +} + +fn jitter() -> f64 { + let mut b = [0u8; 4]; + let _ = getrandom::fill(&mut b); + f64::from(u32::from_le_bytes(b)) / f64::from(u32::MAX) +} + +/// A live connection: the client, the RPC methods, and what the node said. +#[derive(Clone)] +pub struct ChainConnection { + pub endpoint: String, + pub client: OnlineClient, + pub rpc: LegacyRpcMethods, +} + +impl ChainConnection { + /// Genesis, runtime version and metadata as the client currently holds + /// them, for offline signing. Refreshed by the runtime updater. + pub fn chain_state(&self) -> ChainState { + let rv = self.client.runtime_version(); + ChainState { + genesis_hash: self.client.genesis_hash(), + spec_version: rv.spec_version, + transaction_version: rv.transaction_version, + metadata: self.client.metadata(), + } + } +} + +/// Connect to one endpoint and check it is the chain the profile describes. +pub async fn connect_endpoint( + profile: &ChainProfile, + endpoint: &str, +) -> Result { + let rpc = tokio::time::timeout(Duration::from_secs(20), RpcClient::from_url(endpoint)) + .await + .map_err(|_| ChainError::Rpc(format!("{endpoint}: connect timed out")))? + .map_err(|e| ChainError::Rpc(format!("{endpoint}: {e}")))?; + let client = tokio::time::timeout( + Duration::from_secs(60), + OnlineClient::::from_rpc_client(rpc.clone()), + ) + .await + .map_err(|_| ChainError::Rpc(format!("{endpoint}: metadata fetch timed out")))? + .map_err(|e| ChainError::Rpc(format!("{endpoint}: {e}")))?; + let observed = format!("0x{}", hex::encode(client.genesis_hash().0)); + if !genesis_matches(profile, &observed) { + return Err(ChainError::Other(format!( + "{endpoint} is not {}: genesis {observed} does not match the profile's {}", + profile.id.0, profile.genesis_hash + ))); + } + Ok(ChainConnection { + endpoint: endpoint.to_owned(), + client, + rpc: LegacyRpcMethods::new(rpc), + }) +} + +#[derive(Debug, Clone, Default)] +struct Observed { + connected: bool, + endpoint: Option, + spec_version: Option, + transaction_version: Option, + finalized_number: Option, + last_head: Option, + last_error: Option, + failures: u32, +} + +struct Shared { + profile: ChainProfile, + connection: RwLock>, + observed: RwLock, +} + +/// Owns the supervisor task for one chain. Cheap to clone; drop every clone +/// to stop it. +#[derive(Clone)] +pub struct ChainManager { + shared: Arc, +} + +impl ChainManager { + /// Start supervising `profile`: connect, follow heads and upgrades, + /// reconnect and fail over on error, forever. Returns immediately. + pub fn start(profile: ChainProfile) -> Self { + let shared = Arc::new(Shared { + profile, + connection: RwLock::new(None), + observed: RwLock::new(Observed::default()), + }); + let weak = Arc::downgrade(&shared); + tokio::spawn(async move { + let mut backoff = Backoff::new(Duration::from_secs(1), Duration::from_secs(60)); + let mut next_endpoint = 0usize; + loop { + let Some(shared) = weak.upgrade() else { return }; + let endpoints = shared.profile.endpoints.clone(); + let endpoint = endpoints[next_endpoint % endpoints.len()].clone(); + next_endpoint += 1; + drop(shared); + + let Some(shared) = weak.upgrade() else { return }; + let result = connect_endpoint(&shared.profile, &endpoint).await; + match result { + Ok(conn) => { + backoff.reset(); + shared.note_connected(&conn); + *shared.connection.write().expect("connection lock") = Some(conn.clone()); + drop(shared); + let reason = follow(&weak, &conn).await; + let Some(shared) = weak.upgrade() else { return }; + shared.note_disconnected(&reason); + *shared.connection.write().expect("connection lock") = None; + // The endpoint that just dropped is tried again first + // only after the others; a flapping node should not + // monopolise the wallet. + } + Err(e) => { + shared.note_disconnected(&e.to_string()); + tracing::warn!(chain = %shared.profile.id.0, %endpoint, error = %e, "connect failed"); + } + } + backoff.fail(); + let delay = backoff.delay(jitter()); + tokio::time::sleep(delay).await; + } + }); + Self { shared } + } + + pub fn profile(&self) -> &ChainProfile { + &self.shared.profile + } + + /// The live connection, or `Disconnected` with the last error. + pub fn connection(&self) -> Result { + self.shared + .connection + .read() + .expect("connection lock") + .clone() + .ok_or_else(|| { + let o = self.shared.observed.read().expect("observed lock"); + ChainError::Disconnected( + o.last_error + .clone() + .unwrap_or_else(|| "not connected yet".into()), + ) + }) + } + + pub fn status(&self) -> ChainStatus { + let o = self.shared.observed.read().expect("observed lock").clone(); + ChainStatus { + chain: self.shared.profile.id.clone(), + connected: o.connected, + endpoint: o.endpoint, + spec_version: o.spec_version, + transaction_version: o.transaction_version, + finalized_number: o.finalized_number, + head_age_seconds: o.last_head.map(|t| t.elapsed().as_secs()), + last_error: o.last_error, + failures: o.failures, + } + } +} + +impl Shared { + fn note_connected(&self, conn: &ChainConnection) { + let rv = conn.client.runtime_version(); + let mut o = self.observed.write().expect("observed lock"); + o.connected = true; + o.endpoint = Some(conn.endpoint.clone()); + o.spec_version = Some(rv.spec_version); + o.transaction_version = Some(rv.transaction_version); + o.last_error = None; + o.failures = 0; + tracing::info!(chain = %self.profile.id.0, endpoint = %conn.endpoint, spec = rv.spec_version, "connected"); + } + + fn note_disconnected(&self, why: &str) { + let mut o = self.observed.write().expect("observed lock"); + o.connected = false; + o.last_error = Some(why.to_owned()); + o.failures = o.failures.saturating_add(1); + } + + fn note_head(&self, number: u64) { + let mut o = self.observed.write().expect("observed lock"); + o.finalized_number = Some(number); + o.last_head = Some(Instant::now()); + } + + fn note_runtime(&self, spec_version: u32, transaction_version: u32) { + let mut o = self.observed.write().expect("observed lock"); + o.spec_version = Some(spec_version); + o.transaction_version = Some(transaction_version); + } +} + +/// Follow finalized heads and runtime upgrades until something breaks. +/// Returns why it stopped. +async fn follow(weak: &std::sync::Weak, conn: &ChainConnection) -> String { + let mut heads = match conn.client.blocks().subscribe_finalized().await { + Ok(s) => s, + Err(e) => return format!("subscribe finalized: {e}"), + }; + // subxt's updater swaps metadata and runtime version inside the client + // when the chain upgrades; we only need to notice and record it. + let updater = conn.client.updater(); + let mut upgrades = match updater.runtime_updates().await { + Ok(s) => s, + Err(e) => return format!("subscribe runtime updates: {e}"), + }; + let stale = Duration::from_secs(120); + loop { + tokio::select! { + head = heads.next() => match head { + Some(Ok(block)) => { + let Some(shared) = weak.upgrade() else { return "manager dropped".into() }; + shared.note_head(u64::from(block.number())); + } + Some(Err(e)) => return format!("finalized stream: {e}"), + None => return "finalized stream ended".into(), + }, + update = upgrades.next() => match update { + Some(Ok(update)) => { + let version = *update.runtime_version(); + match updater.apply_update(update) { + Ok(()) => { + let Some(shared) = weak.upgrade() else { return "manager dropped".into() }; + shared.note_runtime(version.spec_version, version.transaction_version); + tracing::info!(spec = version.spec_version, "runtime upgraded"); + } + Err(e) => tracing::warn!(error = ?e, "runtime update not applied"), + } + } + Some(Err(e)) => return format!("runtime updates stream: {e}"), + None => return "runtime updates stream ended".into(), + }, + _ = tokio::time::sleep(stale) => return format!("no finalized head for {}s", stale.as_secs()), + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn backoff_doubles_caps_and_jitters_within_half_to_full() { + let mut b = Backoff::new(Duration::from_secs(1), Duration::from_secs(30)); + assert_eq!(b.delay(1.0), Duration::from_secs(1)); + b.fail(); + assert_eq!(b.delay(1.0), Duration::from_secs(2)); + assert_eq!(b.delay(0.0), Duration::from_secs(1)); + for _ in 0..10 { + b.fail(); + } + assert_eq!(b.delay(1.0), Duration::from_secs(30), "capped"); + assert_eq!(b.delay(0.0), Duration::from_secs(15), "half at zero jitter"); + b.reset(); + assert_eq!(b.delay(0.5), Duration::from_millis(750)); + } + + #[test] + fn backoff_never_overflows() { + let mut b = Backoff::new(Duration::from_secs(1), Duration::from_secs(60)); + for _ in 0..100 { + b.fail(); + } + assert_eq!(b.delay(1.0), Duration::from_secs(60)); + } + + /// Live: a dead first endpoint is skipped for the working second one, + /// and a wrong genesis is refused by name. + #[tokio::test] + #[ignore = "talks to a1-planck.quantus.cat"] + async fn fails_over_past_a_dead_endpoint_and_refuses_the_wrong_chain() { + let registry = wallet_core::profiles::ProfileRegistry::shipped().unwrap(); + let mut planck = registry + .get(&wallet_entities::ChainId("planck".into())) + .unwrap() + .clone(); + planck.endpoints = vec![ + "ws://127.0.0.1:1".into(), + "wss://a1-planck.quantus.cat".into(), + ]; + let manager = ChainManager::start(planck.clone()); + let deadline = Instant::now() + Duration::from_secs(90); + while !manager.status().connected && Instant::now() < deadline { + tokio::time::sleep(Duration::from_millis(500)).await; + } + let status = manager.status(); + eprintln!("{status:?}"); + assert!(status.connected, "{status:?}"); + assert_eq!( + status.endpoint.as_deref(), + Some("wss://a1-planck.quantus.cat") + ); + assert_eq!(status.spec_version, Some(148)); + assert!(status.failures == 0); + let state = manager.connection().unwrap().chain_state(); + assert_eq!(state.spec_version, 148); + + // Mainnet's profile pointed at Planck's node: refused, by name. + let mut wrong = registry + .get(&wallet_entities::ChainId("quantus".into())) + .unwrap() + .clone(); + wrong.endpoints = vec!["wss://a1-planck.quantus.cat".into()]; + let err = match connect_endpoint(&wrong, &wrong.endpoints[0]).await { + Ok(_) => panic!("mainnet profile accepted a Planck node"), + Err(e) => e.to_string(), + }; + assert!(err.contains("is not quantus"), "{err}"); + } +} diff --git a/crates/wallet-data/src/substrate/mod.rs b/crates/wallet-data/src/substrate/mod.rs index ee732b3..cfdc909 100644 --- a/crates/wallet-data/src/substrate/mod.rs +++ b/crates/wallet-data/src/substrate/mod.rs @@ -4,7 +4,9 @@ //! the chain profile's context and never by subxt itself. pub mod config; +pub mod connection; pub mod offline; pub use config::{DilithiumSignature, QuantusConfig}; +pub use connection::{ChainConnection, ChainManager}; pub use offline::{ChainState, SignedExtrinsic, sign_call}; diff --git a/crates/wallet-entities/src/lib.rs b/crates/wallet-entities/src/lib.rs index 26e628b..82829ed 100644 --- a/crates/wallet-entities/src/lib.rs +++ b/crates/wallet-entities/src/lib.rs @@ -57,6 +57,26 @@ pub struct ChainProfile { pub signing_context: SigningContextRule, } +/// The wallet's view of its connection to one chain. Everything here is +/// observed, not configured; the UI shows it and decides nothing from it. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, TS)] +#[ts(export)] +pub struct ChainStatus { + pub chain: ChainId, + pub connected: bool, + /// The endpoint currently in use, or the last one tried. + pub endpoint: Option, + pub spec_version: Option, + pub transaction_version: Option, + pub finalized_number: Option, + /// Seconds since the last finalized head arrived; `None` when never. + pub head_age_seconds: Option, + /// The last error, when disconnected. + pub last_error: Option, + /// How many consecutive attempts have failed. + pub failures: u32, +} + /// What kind of thing an asset is on its chain. #[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize, TS)] #[serde(tag = "kind", content = "id", rename_all = "snake_case")] diff --git a/ui/src/api/generated/ChainStatus.ts b/ui/src/api/generated/ChainStatus.ts new file mode 100644 index 0000000..e843f94 --- /dev/null +++ b/ui/src/api/generated/ChainStatus.ts @@ -0,0 +1,24 @@ +// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually. +import type { ChainId } from "./ChainId"; + +/** + * The wallet's view of its connection to one chain. Everything here is + * observed, not configured; the UI shows it and decides nothing from it. + */ +export type ChainStatus = { chain: ChainId, connected: boolean, +/** + * The endpoint currently in use, or the last one tried. + */ +endpoint: string | null, spec_version: number | null, transaction_version: number | null, finalized_number: bigint | null, +/** + * Seconds since the last finalized head arrived; `None` when never. + */ +head_age_seconds: bigint | null, +/** + * The last error, when disconnected. + */ +last_error: string | null, +/** + * How many consecutive attempts have failed. + */ +failures: number, }; diff --git a/ui/src/api/wallet.ts b/ui/src/api/wallet.ts index 964c568..733ea2f 100644 --- a/ui/src/api/wallet.ts +++ b/ui/src/api/wallet.ts @@ -5,7 +5,9 @@ import { invoke } from '@tauri-apps/api/core' import type { AppInfo } from './generated/AppInfo' +import type { ChainId } from './generated/ChainId' import type { ChainProfile } from './generated/ChainProfile' +import type { ChainStatus } from './generated/ChainStatus' import type { CreationStart } from './generated/CreationStart' import type { SessionStatusInfo } from './generated/SessionStatusInfo' import type { SignatureScheme } from './generated/SignatureScheme' @@ -107,3 +109,11 @@ export function sessionStatus(): Promise { export function chainsList(): Promise { return command('chains_list') } + +export function chainConnect(chain: ChainId): Promise { + return command('chain_connect', { chain }) +} + +export function chainStatus(chain: ChainId): Promise { + return command('chain_status', { chain }) +}