Merge pull request #239 from opentensor/release/1.5.12

Release/1.5.12

Author: thewhaleking

.github/workflows/unit-and-integration-test.yml

@@ -48,12 +48,12 @@ jobs:
           PYTHONUNBUFFERED: "1"
         run: |
           source venv/bin/activate
-          python -m uv run pytest -n 2 tests/unit_tests/ --reruns 3
+          python -m uv run pytest tests/unit_tests/ --reruns 3 -s
 
       - name: Integration tests
         timeout-minutes: 20
         env:
           PYTHONUNBUFFERED: "1"
         run: |
           source venv/bin/activate
-          python -m uv run pytest -n 2 tests/integration_tests/ --reruns 3
+          python -m uv run pytest tests/integration_tests/ --reruns 3 -s --forked

CHANGELOG.md

@@ -1,4 +1,11 @@
 # Changelog
+## 1.5.12 /2025-11-167
+* RecursionError in `_wait_with_activity_timeout` with concurrent tasks by @Arthurdw in https://github.com/opentensor/async-substrate-interface/pull/238
+* Improved Test Running + Race Condition Catch by @thewhaleking in https://github.com/opentensor/async-substrate-interface/pull/236
+
+
+**Full Changelog**: https://github.com/opentensor/async-substrate-interface/compare/v1.5.11...v1.5.12
+
 ## 1.5.11 /2025-11-14
 * Race Condition Bug fixes by @thewhaleking in https://github.com/opentensor/async-substrate-interface/pull/234
 

async_substrate_interface/async_substrate.py

@@ -694,9 +694,17 @@ async def _cancel(self):
 
     async def connect(self, force=False):
         if not force:
-            await self._lock.acquire()
+            async with self._lock:
+                return await self._connect_internal(force)
         else:
             logger.debug("Proceeding without acquiring lock.")
+            return await self._connect_internal(force)
+
+    async def _connect_internal(self, force):
+        # Check state again after acquiring lock to avoid duplicate connections
+        if not force and self.state in (State.OPEN, State.CONNECTING):
+            return None
+
         logger.debug(f"Websocket connecting to {self.ws_url}")
         if self._sending is None or self._sending.empty():
             self._sending = asyncio.Queue()
@@ -725,17 +733,13 @@ async def connect(self, force=False):
             except socket.gaierror:
                 logger.debug(f"Hostname not known (this is just for testing")
                 await asyncio.sleep(10)
-                if self._lock.locked():
-                    self._lock.release()
                 return await self.connect(force=force)
             logger.debug("Connection established")
             self.ws = connection
             if self._send_recv_task is None or self._send_recv_task.done():
                 self._send_recv_task = asyncio.get_running_loop().create_task(
                     self._handler(self.ws)
                 )
-        if self._lock.locked():
-            self._lock.release()
         return None
 
     async def _handler(self, ws: ClientConnection) -> Union[None, Exception]:
@@ -838,9 +842,15 @@ async def _exit_with_timer(self):
         """
         try:
             if self.shutdown_timer is not None:
+                logger.debug("Exiting with timer")
                 await asyncio.sleep(self.shutdown_timer)
-            logger.debug("Exiting with timer")
-            await self.shutdown()
+            if (
+                self.state != State.CONNECTING
+                and self._sending.qsize() == 0
+                and not self._received_subscriptions
+                and self._waiting_for_response <= 0
+            ):
+                await self.shutdown()
         except asyncio.CancelledError:
             pass
 
@@ -981,6 +991,7 @@ async def unsubscribe(
             original_id = get_next_id()
             while original_id in self._in_use_ids:
                 original_id = get_next_id()
+            logger.debug(f"Unwatched extrinsic subscription {subscription_id}")
             self._received_subscriptions.pop(subscription_id, None)
 
         to_send = {
@@ -2512,6 +2523,7 @@ async def _make_rpc_request(
         subscription_added = False
 
         async with self.ws as ws:
+            await ws.mark_waiting_for_response()
             for payload in payloads:
                 item_id = await ws.send(payload["payload"])
                 request_manager.add_request(item_id, payload["id"])
@@ -2523,7 +2535,6 @@ async def _make_rpc_request(
                 logger.debug(
                     f"Submitted payload ID {payload['id']} with websocket ID {item_id}: {output_payload}"
                 )
-            await ws.mark_waiting_for_response()
 
             while True:
                 for item_id in request_manager.unresponded():

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "async-substrate-interface"
-version = "1.5.11"
+version = "1.5.12"
 description = "Asyncio library for interacting with substrate. Mostly API-compatible with py-substrate-interface"
 readme = "README.md"
 license = { file = "LICENSE" }
@@ -56,5 +56,6 @@ dev = [
     "pytest-split==0.10.0",
     "pytest-xdist==3.6.1",
     "pytest-rerunfailures==10.2",
-    "bittensor-wallet>=4.0.0"
+    "bittensor-wallet>=4.0.0",
+    "pytest-forked"
 ]

tests/helpers/proxy_server.py

@@ -9,14 +9,21 @@
 
 
 class ProxyServer:
-    def __init__(self, upstream: str, time_til_pause: float, time_til_resume: float):
+    def __init__(
+        self,
+        upstream: str,
+        time_til_pause: float,
+        time_til_resume: float,
+        port: int = 8080,
+    ):
         self.upstream_server = upstream
         self.time_til_pause = time_til_pause
         self.time_til_resume = time_til_resume
         self.upstream_connection = None
         self.connection_time = 0
         self.shutdown_time = 0
         self.resume_time = 0
+        self.port = port
 
     def connect(self):
         self.upstream_connection = connect(self.upstream_server)
@@ -41,7 +48,7 @@ def proxy_request(self, websocket: ServerConnection):
             websocket.send(recd)
 
     def serve(self):
-        with serve(self.proxy_request, "localhost", 8080) as self.server:
+        with serve(self.proxy_request, "localhost", self.port) as self.server:
             self.server.serve_forever()
 
     def connect_and_serve(self):

tests/integration_tests/test_async_substrate_interface.py

@@ -3,6 +3,7 @@
 import os.path
 import time
 import threading
+import socket
 
 import bittensor_wallet
 import pytest
@@ -195,8 +196,16 @@ async def test_query_map_with_odd_number_of_params():
     print("test_query_map_with_odd_number_of_params succeeded")
 
 
+@pytest.mark.skip("Weird issue with the GitHub Actions runner")
 @pytest.mark.asyncio
 async def test_improved_reconnection():
+    def get_free_port():
+        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
+            s.bind(("", 0))  # Bind to port 0 = OS picks free port
+            s.listen(1)
+            port_ = s.getsockname()[1]
+        return port_
+
     print("Testing test_improved_reconnection")
     ws_logger_path = "/tmp/websockets-proxy-test"
     ws_logger = logging.getLogger("websockets.proxy")
@@ -210,14 +219,15 @@ async def test_improved_reconnection():
         os.remove(asi_logger_path)
     logger.setLevel(logging.DEBUG)
     logger.addHandler(logging.FileHandler(asi_logger_path))
+    port = get_free_port()
+    print(f"Testing using server on port {port}")
+    proxy = ProxyServer("wss://archive.sub.latent.to", 10, 20, port=port)
 
-    proxy = ProxyServer("wss://archive.sub.latent.to", 10, 20)
-
-    server_thread = threading.Thread(target=proxy.connect_and_serve)
+    server_thread = threading.Thread(target=proxy.connect_and_serve, daemon=True)
     server_thread.start()
     await asyncio.sleep(3)  # give the server start up time
     async with AsyncSubstrateInterface(
-        "ws://localhost:8080",
+        f"ws://localhost:{port}",
         ss58_format=42,
         chain_name="Bittensor",
         retry_timeout=10.0,
@@ -247,7 +257,7 @@ async def test_improved_reconnection():
             assert "Pausing" in f.read()
         with open(asi_logger_path, "r") as f:
             assert "Timeout/ConnectionClosed occurred." in f.read()
-    shutdown_thread = threading.Thread(target=proxy.close)
+    shutdown_thread = threading.Thread(target=proxy.close, daemon=True)
     shutdown_thread.start()
     shutdown_thread.join(timeout=5)
     server_thread.join(timeout=5)
@@ -293,3 +303,45 @@ async def test_get_payment_info():
         assert partial_fee_all_options > partial_fee_no_era
         assert partial_fee_all_options > partial_fee_era
     print("test_get_payment_info succeeded")
+
+
+@pytest.mark.asyncio
+async def test_concurrent_rpc_requests():
+    """
+    Test that multiple concurrent RPC requests on a shared connection work correctly.
+
+    This test verifies the fix for the issue where multiple concurrent tasks
+    re-initializing the WebSocket connection caused requests to hang.
+    """
+    print("Testing test_concurrent_rpc_requests")
+
+    async def concurrent_task(substrate_, task_id):
+        """Make multiple RPC calls from a single task."""
+        for i in range(5):
+            result = await substrate_.get_block_number(None)
+            assert isinstance(result, int)
+            assert result > 0
+
+    async with AsyncSubstrateInterface(LATENT_LITE_ENTRYPOINT) as substrate:
+        # Run 5 concurrent tasks, each making 5 RPC calls (25 total)
+        # This tests that the connection is properly shared without re-initialization
+        tasks = [concurrent_task(substrate, i) for i in range(5)]
+        await asyncio.gather(*tasks)
+
+    print("test_concurrent_rpc_requests succeeded")
+
+
+@pytest.mark.asyncio
+async def test_wait_for_block():
+    async def handler(_):
+        return True
+
+    substrate = AsyncSubstrateInterface(
+        LATENT_LITE_ENTRYPOINT, ss58_format=42, chain_name="Bittensor"
+    )
+    await substrate.initialize()
+    current_block = await substrate.get_block_number(None)
+    result = await substrate.wait_for_block(
+        current_block + 3, result_handler=handler, task_return=False
+    )
+    assert result is True