feat(substrate): a supervised connection per chain, failing over and following upgrades

`ChainManager::start(profile)` spawns a supervisor that tries the profile's
endpoints in order, refuses a node whose genesis is not the profile's (by
name, so a typo in an override file cannot quietly sign for the wrong
chain), records the runtime version at connect, follows finalized heads,
applies subxt's runtime updates as the chain upgrades, and on any failure
rotates to the next endpoint after an exponential backoff with full jitter,
capped at a minute. A head that fails to arrive for two minutes counts as
a failure. `status()` reports the endpoint in use, spec and transaction
version, finalized height, head age, the last error and the failure count.

Backoff is a pure type with the schedule pinned in tests. The live test
gives Planck's profile a dead first endpoint and the real second one: it
connects to the second at spec 148 with zero failures, and mainnet's
profile pointed at Planck's node is refused with "is not quantus".

Supervisors start on demand from `chain_connect`, never at boot, because a
profile whose endpoint does not exist yet (#53) would otherwise retry
forever for nobody. `chain_status` reads without starting anything.

Not yet exercised: a stream dropping mid-session and a runtime upgrade
arriving while connected. Both go through the same loop and will be
driven on the dev node in #24.

Closes #20

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_014ftBXYuba8ARhQeF74oUgW

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<SessionStatusInfo, W
 pub fn chains_list(state: State<'_, AppState>) -> Result<Vec<ChainProfile>, 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<ChainStatus, WalletError> {
+    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<ChainStatus, WalletError> {
+    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,
+        }),
+    }
+}

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");

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<Option<PendingCreation>>,
     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<HashMap<ChainId, ChainManager>>,
 }
 
 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()),
         })
     }
 }

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

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<QuantusConfig>,
+    pub rpc: LegacyRpcMethods<QuantusConfig>,
+}
+
+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<ChainConnection, ChainError> {
+    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::<QuantusConfig>::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<String>,
+    spec_version: Option<u32>,
+    transaction_version: Option<u32>,
+    finalized_number: Option<u64>,
+    last_head: Option<Instant>,
+    last_error: Option<String>,
+    failures: u32,
+}
+
+struct Shared {
+    profile: ChainProfile,
+    connection: RwLock<Option<ChainConnection>>,
+    observed: RwLock<Observed>,
+}
+
+/// Owns the supervisor task for one chain. Cheap to clone; drop every clone
+/// to stop it.
+#[derive(Clone)]
+pub struct ChainManager {
+    shared: Arc<Shared>,
+}
+
+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<ChainConnection, ChainError> {
+        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<Shared>, 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}");
+    }
+}

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};

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<String>,
+    pub spec_version: Option<u32>,
+    pub transaction_version: Option<u32>,
+    pub finalized_number: Option<u64>,
+    /// Seconds since the last finalized head arrived; `None` when never.
+    pub head_age_seconds: Option<u64>,
+    /// The last error, when disconnected.
+    pub last_error: Option<String>,
+    /// 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")]

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, };

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<SessionStatusInfo> {
 export function chainsList(): Promise<ChainProfile[]> {
   return command<ChainProfile[]>('chains_list')
 }
+
+export function chainConnect(chain: ChainId): Promise<ChainStatus> {
+  return command<ChainStatus>('chain_connect', { chain })
+}
+
+export function chainStatus(chain: ChainId): Promise<ChainStatus> {
+  return command<ChainStatus>('chain_status', { chain })
+}