fix: support insecure ws:// URLs + auto-reconnect Bittensor client

- Vendor bittensor-rs and patch BittensorClient::new to use
  from_insecure_url, allowing both ws:// and wss:// endpoints
- BlockSync now recreates the BittensorClient from scratch when the
  broadcast channel closes (dead websocket), with exponential backoff
- SubtensorClient gains reconnect() and with_reconnect() methods for
  automatic transport-error recovery on metagraph sync and weight ops

Author: factorydroid

Cargo.lock

@@ -35,9 +35,8 @@ authors = ["Platform Network"]
 license = "Apache-2.0"
 
 [workspace.dependencies]
-# Bittensor (with CRv4 timelock encryption support and new Subtensor API)
-# Updated to d2ca7c2: fix epoch calculation to match subtensor formula
-bittensor-rs = { git = "https://github.com/CortexLM/bittensor-rs", rev = "99b6325" }
+# Bittensor (vendored with insecure-URL fix: ws:// support + auto-reconnect)
+bittensor-rs = { path = "vendor/bittensor-rs" }
 
 # Async runtime
 tokio = { version = "1.40", features = ["full", "sync", "macros", "rt-multi-thread"] }

Cargo.toml

@@ -65,6 +65,8 @@ pub struct BlockSync {
     config: BlockSyncConfig,
     listener: Option<BlockListener>,
     client: Option<Arc<BittensorClient>>,
+    /// Stored so we can recreate the client on disconnect.
+    rpc_url: Option<String>,
     running: Arc<RwLock<bool>>,
     event_tx: mpsc::Sender<BlockSyncEvent>,
     event_rx: Option<mpsc::Receiver<BlockSyncEvent>>,
@@ -83,6 +85,7 @@ impl BlockSync {
             config,
             listener: None,
             client: None,
+            rpc_url: None,
             running: Arc::new(RwLock::new(false)),
             event_tx,
             event_rx: Some(event_rx),
@@ -129,6 +132,7 @@ impl BlockSync {
             epoch_info.blocks_remaining, mins, secs
         );
 
+        self.rpc_url = Some(client.rpc_url.clone());
         self.listener = Some(listener);
         self.client = Some(client);
 
@@ -140,7 +144,12 @@ impl BlockSync {
         *self.tempo.read().await
     }
 
-    /// Start the block sync loop
+    /// Start the block sync loop.
+    ///
+    /// The spawned background task monitors the block listener and, if the
+    /// underlying Bittensor RPC connection dies permanently (broadcast channel
+    /// closed), it will **recreate** the `BittensorClient` from scratch,
+    /// re-initialise the `BlockListener`, and resume processing blocks.
     pub async fn start(&self) -> anyhow::Result<()> {
         let listener = self
             .listener
@@ -152,6 +161,11 @@ impl BlockSync {
             .as_ref()
             .ok_or_else(|| anyhow::anyhow!("Not connected"))?;
 
+        let rpc_url = self
+            .rpc_url
+            .clone()
+            .ok_or_else(|| anyhow::anyhow!("RPC URL not set"))?;
+
         // Check if already running
         {
             let mut running = self.running.write().await;
@@ -162,28 +176,32 @@ impl BlockSync {
         }
 
         // Subscribe to block events
-        let mut block_rx = listener.subscribe();
+        let block_rx = listener.subscribe();
         let event_tx = self.event_tx.clone();
         let running = self.running.clone();
         let current_block = self.current_block.clone();
         let current_epoch = self.current_epoch.clone();
         let current_phase = self.current_phase.clone();
+        let config = self.config.clone();
 
         // Start the listener
         listener.start(client.clone()).await?;
 
-        // Process events in background
+        // Process events in background with automatic client recreation
         tokio::spawn(async move {
             let mut was_disconnected = false;
             let mut first_block_seen = false;
+            let mut current_block_rx = block_rx;
+            let mut consecutive_reconnect_failures: u32 = 0;
 
             loop {
                 if !*running.read().await {
                     break;
                 }
 
-                match block_rx.recv().await {
+                match current_block_rx.recv().await {
                     Ok(event) => {
+                        consecutive_reconnect_failures = 0;
                         let should_break = BlockSync::handle_block_event(
                             event,
                             &event_tx,
@@ -203,8 +221,62 @@ impl BlockSync {
                         warn!("Block sync lagged by {} events", n);
                     }
                     Err(broadcast::error::RecvError::Closed) => {
-                        info!("Block event channel closed");
-                        break;
+                        // The BlockListener's internal task has died and the
+                        // broadcast channel is gone.  The old BittensorClient's
+                        // websocket is likely dead too, so retrying with it
+                        // will not help.  Recreate everything from scratch.
+
+                        if !*running.read().await {
+                            break;
+                        }
+
+                        consecutive_reconnect_failures += 1;
+                        let delay_secs = std::cmp::min(5 * consecutive_reconnect_failures, 60);
+                        warn!(
+                            attempt = consecutive_reconnect_failures,
+                            delay_secs,
+                            "Block listener channel closed — recreating Bittensor client in {}s",
+                            delay_secs,
+                        );
+
+                        let _ = event_tx
+                            .send(BlockSyncEvent::Disconnected(
+                                "Block listener channel closed, recreating client".into(),
+                            ))
+                            .await;
+
+                        tokio::time::sleep(std::time::Duration::from_secs(delay_secs as u64))
+                            .await;
+
+                        match BlockSync::recreate_listener(&rpc_url, &config).await {
+                            Ok((new_client, new_listener, new_rx, epoch_info)) => {
+                                info!(
+                                    block = epoch_info.current_block,
+                                    epoch = epoch_info.epoch_number,
+                                    "Bittensor client recreated successfully"
+                                );
+                                *current_block.write().await = epoch_info.current_block;
+                                *current_epoch.write().await = epoch_info.epoch_number;
+                                *current_phase.write().await = epoch_info.phase;
+                                current_block_rx = new_rx;
+                                was_disconnected = true;
+                                consecutive_reconnect_failures = 0;
+
+                                if let Err(e) = new_listener.start(new_client).await {
+                                    warn!("Failed to start recreated listener: {}", e);
+                                } else {
+                                    let _ =
+                                        event_tx.send(BlockSyncEvent::Reconnected).await;
+                                }
+                            }
+                            Err(e) => {
+                                warn!(
+                                    attempt = consecutive_reconnect_failures,
+                                    error = %e,
+                                    "Failed to recreate Bittensor client, will retry"
+                                );
+                            }
+                        }
                     }
                 }
             }
@@ -213,6 +285,34 @@ impl BlockSync {
         Ok(())
     }
 
+    /// Create a fresh `BittensorClient` + `BlockListener` from the RPC URL.
+    async fn recreate_listener(
+        rpc_url: &str,
+        config: &BlockSyncConfig,
+    ) -> anyhow::Result<(
+        Arc<BittensorClient>,
+        BlockListener,
+        broadcast::Receiver<BlockEvent>,
+        EpochInfo,
+    )> {
+        let client = Arc::new(BittensorClient::new(rpc_url).await?);
+
+        let listener_config = BlockListenerConfig {
+            netuid: config.netuid,
+            channel_capacity: config.channel_capacity,
+            auto_reconnect: true,
+            reconnect_delay_ms: 5000,
+        };
+
+        let listener = BlockListener::new(listener_config);
+        listener.init(&client).await?;
+
+        let epoch_info = listener.current_epoch_info(&client).await?;
+        let rx = listener.subscribe();
+
+        Ok((client, listener, rx, epoch_info))
+    }
+
     async fn handle_block_event(
         event: BlockEvent,
         event_tx: &mpsc::Sender<BlockSyncEvent>,

crates/bittensor-integration/src/block_sync.rs

@@ -39,6 +39,61 @@ impl SubtensorClient {
         Ok(())
     }
 
+    /// Reconnect to Subtensor by creating a fresh client.
+    /// Useful when the underlying websocket has died.
+    pub async fn reconnect(&mut self) -> Result<()> {
+        info!(
+            "Reconnecting to Subtensor: {}",
+            self.config.endpoint
+        );
+        self.client = None;
+        let client = BittensorClient::new(&self.config.endpoint).await?;
+        self.client = Some(client);
+        info!("Reconnected to Subtensor");
+        Ok(())
+    }
+
+    /// Execute a fallible operation, reconnecting once if the error looks
+    /// like a dead websocket / RPC transport failure.
+    pub async fn with_reconnect<F, Fut, T>(&mut self, op: F) -> Result<T>
+    where
+        F: Fn(&BittensorClient) -> Fut,
+        Fut: std::future::Future<Output = Result<T>>,
+    {
+        let client = self
+            .client
+            .as_ref()
+            .ok_or_else(|| anyhow::anyhow!("Not connected to Subtensor"))?;
+
+        match op(client).await {
+            Ok(val) => Ok(val),
+            Err(first_err) => {
+                let msg = first_err.to_string().to_lowercase();
+                let is_transport = msg.contains("websocket")
+                    || msg.contains("connection reset")
+                    || msg.contains("connection refused")
+                    || msg.contains("broken pipe")
+                    || msg.contains("eof")
+                    || msg.contains("channel closed")
+                    || msg.contains("timed out")
+                    || msg.contains("timeout");
+
+                if !is_transport {
+                    return Err(first_err);
+                }
+
+                info!("Transport error detected, attempting reconnect: {}", first_err);
+                self.reconnect().await?;
+
+                let client = self
+                    .client
+                    .as_ref()
+                    .ok_or_else(|| anyhow::anyhow!("Not connected after reconnect"))?;
+                op(client).await
+            }
+        }
+    }
+
     /// Set the signer from a seed phrase or key
     pub fn set_signer(&mut self, seed: &str) -> Result<()> {
         let signer = signer_from_seed(seed)?;

crates/bittensor-integration/src/client.rs

@@ -0,0 +1,44 @@
+[package]
+name = "bittensor-rs"
+version = "0.1.0"
+edition = "2021"
+authors = ["Cortex Foundation"]
+description = "A Rust SDK for interacting with the Bittensor network"
+repository = "https://github.com/CortexLM/bittensor-rs"
+readme = "README.md"
+license = "MIT"
+keywords = ["bittensor", "blockchain", "substrate", "sdk"]
+categories = ["cryptography::cryptocurrencies", "network-programming"]
+
+[dependencies]
+anyhow = "1.0.100"
+async-trait = "0.1.89"
+futures = "0.3.31"
+hex = "0.4.3"
+num-traits = "0.2.19"
+parity-scale-codec = { version = "3.7.5", features = ["derive"] }
+scale-decode = "0.16.0"
+scale-encode = "0.10.0"
+scale-info = "2.11"
+serde = { version = "1.0.228", features = ["derive"] }
+serde_json = "1.0"
+sp-core = "38.1.0"
+sp-runtime = "44.0.0"
+subxt = "0.44"
+thiserror = "2.0.17"
+tokio = { version = "1.48.0", features = ["full"] }
+rand = "0.9"
+regex = "1.10"
+tracing = "0.1"
+chrono = { version = "0.4", features = ["serde"] }
+
+# CRv4 / Timelock encryption dependencies (same as subtensor)
+tle = { git = "https://github.com/ideal-lab5/timelock", rev = "5416406cfd32799e31e1795393d4916894de4468", default-features = false, features = ["std"] }
+w3f-bls = { git = "https://github.com/opentensor/bls", branch = "fix-no-std", default-features = false, features = ["std"] }
+ark-serialize = { version = "0.4.0", features = ["derive"] }
+ark-std = "0.4.0"
+sha2 = "0.10"
+rand_chacha = "0.3"
+
+[patch.crates-io]
+w3f-bls = { git = "https://github.com/opentensor/bls", branch = "fix-no-std" }