explorer/python: Reconnect to RPC if connection is unavailable/lost

bin/explorer/python/explorer.py

@@ -17,9 +17,9 @@
 # along with this program.  If not, see <https://www.gnu.org/licenses/>.
 from datetime import datetime, timezone
 
-from quart import Quart, render_template, abort, request, redirect, url_for
+from quart import Quart, render_template, abort, request, redirect, url_for, Response
 
-from rpc_client import JsonRpcPool, JsonRpcError
+from rpc_client import JsonRpcPool, JsonRpcError, RpcUnavailableError
 
 app = Quart(__name__)
 app.config.update(
@@ -27,6 +27,7 @@ app.config.update(
     RPC_PORT="22222",
     RPC_MIN_CONNECTIONS=5,
     RPC_MAX_CONNECTIONS=50,
+    RPC_RECONNECT_INTERVAL=5.0,
     NETWORK="Testnet",
 )
 
@@ -42,9 +43,10 @@ async def startup():
         port=app.config["RPC_PORT"],
         min_connections=app.config["RPC_MIN_CONNECTIONS"],
         max_connections=app.config["RPC_MAX_CONNECTIONS"],
+        reconnect_interval=app.config["RPC_RECONNECT_INTERVAL"],
     )
     await rpc.start()
-    app.logger.info(f"RPC pool started: {app.config['RPC_HOST']}:{app.config['RPC_PORT']}")
+    app.logger.info(f"RPC pool initialized for {app.config['RPC_HOST']}:{app.config['RPC_PORT']}")
 
 
 @app.after_serving
@@ -71,9 +73,20 @@ async def handle_rpc_error(error: JsonRpcError):
     ), 500
 
 
+@app.errorhandler(RpcUnavailableError)
+async def handle_rpc_unavailable(error: RpcUnavailableError):
+    app.logger.error(f"RPC Unavailable: {error}")
+    return await render_template(
+        "error.html",
+        network=app.config["NETWORK"],
+        error_code="503",
+        error="Blockchain node is currently unavailable. Please try again later."
+    ), 503
+
+
 @app.errorhandler(ConnectionError)
 async def handle_connection_error(error):
-    app.logger.error(f"RPC Connection Error: {error}")
+    app.logger.error(f"Connection Error: {error}")
     return await render_template(
         "error.html",
         network=app.config["NETWORK"],
@@ -106,6 +119,16 @@ def format_hashrate(hashrate: float) -> str:
         return f"{hashrate:.2f} H/s"
 
 
+def format_bytes(size: int) -> str:
+    """Format byte size with appropriate unit."""
+    if size >= 1024 * 1024:
+        return f"{size / (1024 * 1024):.2f} MB"
+    elif size >= 1024:
+        return f"{size / 1024:.2f} KB"
+    else:
+        return f"{size} bytes"
+
+
 @app.route("/")
 async def index():
     current_difficulty = await rpc.call("current_difficulty", params=[])
@@ -212,16 +235,6 @@ async def search():
     ), 404
 
 
-def format_bytes(size: int) -> str:
-    """Format byte size with appropriate unit."""
-    if size >= 1024 * 1024:
-        return f"{size / (1024 * 1024):.2f} MB"
-    elif size >= 1024:
-        return f"{size / 1024:.2f} KB"
-    else:
-        return f"{size} bytes"
-
-
 @app.route("/contract/<contract_id>")
 async def get_contract(contract_id: str):
     current_difficulty = await rpc.call("current_difficulty", params=[])
@@ -288,7 +301,6 @@ async def daily_tx_chart():
     matplotlib.use('Agg')  # Non-interactive backend
     import matplotlib.pyplot as plt
     import matplotlib.dates as mdates
-    from datetime import datetime, timezone
 
     stats_data = await rpc.call("get_stats", params=[])
     daily_stats = stats_data.get("daily_stats", [])
@@ -338,7 +350,6 @@ async def daily_tx_chart():
     plt.close(fig)
     buf.seek(0)
 
-    from quart import Response
     return Response(buf.getvalue(), mimetype='image/png')
 
 

bin/explorer/python/rpc_client.py

@@ -1,11 +1,14 @@
 import asyncio
 import json
-from dataclasses import dataclass
-from typing import Any
+import logging
+from typing import Any, Optional
 from contextlib import asynccontextmanager
 
+logger = logging.getLogger(__name__)
+
 
 class JsonRpcError(Exception):
+    """Error returned by the RPC server."""
     def __init__(self, code: int, message: str, data: Any = None):
         self.code = code
         self.message = message
@@ -13,8 +16,13 @@ class JsonRpcError(Exception):
         super().__init__(f"RPC Error {code}: {message}")
 
 
+class RpcUnavailableError(Exception):
+    """Raised when RPC endpoint is not reachable."""
+    pass
+
+
 class JsonRpcConnection:
-    """Single JSON-RPC connection over TCP"""
+    """Single JSON-RPC connection over TCP."""
 
     def __init__(self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter):
         self.reader = reader
@@ -84,7 +92,13 @@ class JsonRpcConnection:
 
 
 class JsonRpcPool:
-    """Connection pool with automatic reconnection"""
+    """
+    Connection pool with background reconnection.
+
+    - If RPC is unavailable, immediately returns error (no waiting)
+    - Background task keeps trying to reconnect every N seconds
+    - Once connected, requests work again
+    """
 
     def __init__(
         self,
@@ -92,84 +106,154 @@ class JsonRpcPool:
         port: int,
         min_connections: int = 5,
         max_connections: int = 20,
+        reconnect_interval: float = 5.0,
+        connect_timeout: float = 5.0,
     ):
         self.host = host
         self.port = port
         self.min_connections = min_connections
         self.max_connections = max_connections
+        self.reconnect_interval = reconnect_interval
+        self.connect_timeout = connect_timeout
 
         self._pool: asyncio.Queue[JsonRpcConnection] = None
         self._semaphore: asyncio.Semaphore = None
         self._connection_count = 0
         self._lock = asyncio.Lock()
         self._closed = False
+        self._available = False
+        self._reconnect_task: Optional[asyncio.Task] = None
 
     async def start(self):
-        """Initialize the pool with minimum connections"""
+        """Initialize the pool"""
         self._pool = asyncio.Queue()
         self._semaphore = asyncio.Semaphore(self.max_connections)
         self._connection_count = 0
+        self._closed = False
 
+        # Try to create initial connections
+        success_count = 0
         for _ in range(self.min_connections):
-            try:
-                conn = await self._create_connection()
+            conn = await self._create_connection()
+            if conn:
                 await self._pool.put(conn)
-            except Exception as e:
-                print(f"Warning: Failed to create initial connection: {e}")
+                success_count += 1
 
-    async def _create_connection(self) -> JsonRpcConnection:
-        reader, writer = await asyncio.open_connection(self.host, self.port)
-        async with self._lock:
-            self._connection_count += 1
-        return JsonRpcConnection(reader, writer)
+        if success_count > 0:
+            self._available = True
+            logger.info(f"RPC pool started with {success_count} connections")
+        else:
+            self._available = False
+            logger.warning(f"RPC {self.host}:{self.port} unavailable, will retry in background")
+            self._start_reconnect_task()
+
+    def _start_reconnect_task(self):
+        """Start background reconnection task if not already running."""
+        if self._reconnect_task is None or self._reconnect_task.done():
+            self._reconnect_task = asyncio.create_task(self._reconnect_loop())
+
+    async def _reconnect_loop(self):
+        """Background task that keeps trying to reconnect."""
+        while not self._closed and not self._available:
+            await asyncio.sleep(self.reconnect_interval)
+
+            if self._closed:
+                break
+
+            conn = await self._create_connection()
+            if conn:
+                await self._pool.put(conn)
+                self._available = True
+                logger.info(f"RPC {self.host}:{self.port} reconnected")
+                break
+            else:
+                logger.debug(f"RPC {self.host}:{self.port} still unavailable, retrying...")
+
+    async def _create_connection(self) -> Optional[JsonRpcConnection]:
+        """Create a new connection. Returns None if connection fails."""
+        try:
+            reader, writer = await asyncio.wait_for(
+                asyncio.open_connection(self.host, self.port),
+                timeout=self.connect_timeout
+            )
+            async with self._lock:
+                self._connection_count += 1
+            return JsonRpcConnection(reader, writer)
+        except (asyncio.TimeoutError, OSError) as e:
+            logger.debug(f"Connection failed: {e}")
+            return None
 
     async def _destroy_connection(self, conn: JsonRpcConnection):
+        """Close and clean up a connection"""
         await conn.close()
         async with self._lock:
-            self._connection_count -= 1
+            self._connection_count = max(0, self._connection_count - 1)
 
-    @asynccontextmanager
-    async def connection(self):
-        """Acquire a connection from the pool"""
+    async def call(self, method: str, params: Any = None, timeout: float = 30.0) -> Any:
+        """Make an RPC call. Raises RpcUnavailableError immediately if not connected."""
         if self._closed:
             raise RuntimeError("Pool's closed")
 
-        conn = None
+        if not self._available:
+            raise RpcUnavailableError(f"RPC {self.host}:{self.port} is unavailable")
 
         async with self._semaphore:
-            # Try to get an existing connection
+            # Get or create connection
+            conn = None
             while not self._pool.empty():
-                conn = await self._pool.get()
-                if not conn.is_closed:
+                try:
+                    conn = self._pool.get_nowait()
+                    if not conn.is_closed:
+                        break
+                    await self._destroy_connection(conn)
+                    conn = None
+                except asyncio.QueueEmpty:
                     break
-                await self._destroy_connection(conn)
-                conn = None
 
-            # Create new if needed
             if conn is None:
                 conn = await self._create_connection()
+                if conn is None:
+                    self._available = False
+                    self._start_reconnect_task()
+                    raise RpcUnavailableError(f"RPC {self.host}:{self.port} is unavailable")
 
+            # Make the call
             try:
-                yield conn
-            except (ConnectionError, TimeoutError):
-                # Connection is bad, don't return to pool
-                await self._destroy_connection(conn)
+                result = await conn.call(method, params, timeout)
+                await self._pool.put(conn)
+                return result
+            except JsonRpcError:
+                # Server error - connection is still good
+                await self._pool.put(conn)
                 raise
-            else:
-                # Return healthy connection to pool
-                if not conn.is_closed:
-                    await self._pool.put(conn)
-                else:
-                    await self._destroy_connection(conn)
+            except (ConnectionError, TimeoutError) as e:
+                # Connection failed
+                await self._destroy_connection(conn)
+                self._available = False
+                self._start_reconnect_task()
+                raise RpcUnavailableError(f"RPC {self.host}:{self.port} is unavailable: {e}")
 
-    async def call(self, method: str, params: Any = None, timeout: float = 30.0) -> Any:
-        """Make an RPC call using a pooled connection"""
-        async with self.connection() as conn:
-            return await conn.call(method, params, timeout)
+    @property
+    def is_available(self) -> bool:
+        """Check if RPC is currently available."""
+        return self._available
 
     async def close(self):
         """Close all connections"""
         self._closed = True
+
+        if self._reconnect_task and not self._reconnect_task.done():
+            self._reconnect_task.cancel()
+            try:
+                await self._reconnect_task
+            except asyncio.CancelledError:
+                pass
+
         while not self._pool.empty():
-            conn = await self._pool.get()
-            await self._destroy_connection(conn)
+            try:
+                conn = self._pool.get_nowait()
+                await self._destroy_connection(conn)
+            except asyncio.QueueEmpty:
+                break
+
+        logger.info("RPC pool closed")

bin/explorer/python/templates/error.html

@@ -9,7 +9,7 @@
 <body>
   <header class="header">
     <div class="container flex justify-between items-center">
-      <div class="logo">>DarkFi {{ network }} Blocks</div>
+      <div class="logo">DarkFi {{ network }} Blocks</div>
       <nav class="nav">
         <a href="/">Explorer</a>
         <a href="/contracts">Contracts</a>