websocket timeouts and retries

Author: merolish

apps/hip-3-pusher/pyproject.toml

@@ -5,17 +5,18 @@ description = "Hyperliquid HIP-3 market oracle pusher"
 readme = "README.md"
 requires-python = "==3.13.*"
 dependencies = [
-    "boto3~=1.40.34",
+    "boto3~=1.40.38",
     "cryptography~=46.0.1",
     "hyperliquid-python-sdk~=0.19.0",
     "loguru~=0.7.3",
     "opentelemetry-exporter-prometheus~=0.58b0",
     "opentelemetry-sdk~=1.37.0",
     "prometheus-client~=0.23.1",
+    "tenacity~=9.1.2",
     "websockets~=15.0.1",
 ]
 
 [dependency-groups]
 dev = [
-    "pytest>=8.4.2",
+    "pytest~=8.4.2",
 ]

apps/hip-3-pusher/src/pusher/config.py

@@ -1,5 +1,7 @@
 from pydantic import BaseModel
 
+STALE_TIMEOUT_SECONDS = 5
+
 
 class KMSConfig(BaseModel):
     enable_kms: bool

apps/hip-3-pusher/src/pusher/exception.py

@@ -0,0 +1,2 @@
+class StaleConnection(Exception):
+    pass

apps/hip-3-pusher/src/pusher/hermes_listener.py

@@ -3,8 +3,10 @@
 from loguru import logger
 import time
 import websockets
+from tenacity import retry, retry_if_exception_type, wait_exponential
 
-from pusher.config import Config
+from pusher.config import Config, STALE_TIMEOUT_SECONDS
+from pusher.exception import StaleConnection
 from pusher.price_state import PriceState, PriceUpdate
 
 
@@ -31,14 +33,13 @@ def get_subscribe_request(self):
     async def subscribe_all(self):
         await asyncio.gather(*(self.subscribe_single(url) for url in self.hermes_urls))
 
+    @retry(
+        retry=retry_if_exception_type((StaleConnection, websockets.ConnectionClosed)),
+        wait=wait_exponential(multiplier=1, min=1, max=10),
+        reraise=True,
+    )
     async def subscribe_single(self, url):
-        while True:
-            try:
-                await self.subscribe_single_inner(url)
-            except websockets.ConnectionClosed:
-                logger.error("Connection to {} closed; retrying", url)
-            except Exception as e:
-                logger.exception("Error on {}: {}", url, e)
+        return await self.subscribe_single_inner(url)
 
     async def subscribe_single_inner(self, url):
         async with websockets.connect(url) as ws:
@@ -48,12 +49,19 @@ async def subscribe_single_inner(self, url):
             logger.info("Sent Hermes subscribe request to {}", url)
 
             # listen for updates
-            async for message in ws:
+            while True:
                 try:
+                    message = await asyncio.wait_for(ws.recv(), timeout=STALE_TIMEOUT_SECONDS)
                     data = json.loads(message)
                     self.parse_hermes_message(data)
+                except asyncio.TimeoutError:
+                    raise StaleConnection(f"No messages in {STALE_TIMEOUT_SECONDS} seconds, reconnecting")
+                except websockets.ConnectionClosed:
+                    raise
                 except json.JSONDecodeError as e:
                     logger.error("Failed to decode JSON message: {}", e)
+                except Exception as e:
+                    logger.error("Unexpected exception: {}", e)
 
     def parse_hermes_message(self, data):
         """

apps/hip-3-pusher/src/pusher/hyperliquid_listener.py

@@ -2,9 +2,11 @@
 import json
 import websockets
 from loguru import logger
+from tenacity import retry, retry_if_exception_type, wait_exponential
 import time
 
-from pusher.config import Config
+from pusher.config import Config, STALE_TIMEOUT_SECONDS
+from pusher.exception import StaleConnection
 from pusher.price_state import PriceState, PriceUpdate
 
 # This will be in config, but note here.
@@ -32,14 +34,13 @@ def get_subscribe_request(self, asset):
     async def subscribe_all(self):
         await asyncio.gather(*(self.subscribe_single(hyperliquid_ws_url) for hyperliquid_ws_url in self.hyperliquid_ws_urls))
 
+    @retry(
+        retry=retry_if_exception_type((StaleConnection, websockets.ConnectionClosed)),
+        wait=wait_exponential(multiplier=1, min=1, max=10),
+        reraise=True,
+    )
     async def subscribe_single(self, url):
-        while True:
-            try:
-                await self.subscribe_single_inner(url)
-            except websockets.ConnectionClosed:
-                logger.error("Connection to {} closed; retrying", url)
-            except Exception as e:
-                logger.exception("Error on {}: {}", url, e)
+        return await self.subscribe_single_inner(url)
 
     async def subscribe_single_inner(self, url):
         async with websockets.connect(url) as ws:
@@ -48,8 +49,9 @@ async def subscribe_single_inner(self, url):
             logger.info("Sent subscribe request to {}", url)
 
             # listen for updates
-            async for message in ws:
+            while True:
                 try:
+                    message = await asyncio.wait_for(ws.recv(), timeout=STALE_TIMEOUT_SECONDS)
                     data = json.loads(message)
                     channel = data.get("channel", None)
                     if not channel:
@@ -62,8 +64,14 @@ async def subscribe_single_inner(self, url):
                         self.parse_hyperliquid_ws_message(data)
                     else:
                         logger.error("Received unknown channel: {}", channel)
+                except asyncio.TimeoutError:
+                    raise StaleConnection(f"No messages in {STALE_TIMEOUT_SECONDS} seconds, reconnecting...")
+                except websockets.ConnectionClosed:
+                    raise
                 except json.JSONDecodeError as e:
                     logger.error("Failed to decode JSON message: {} error: {}", message, e)
+                except Exception as e:
+                    logger.error("Unexpected exception: {}", e)
 
     def parse_hyperliquid_ws_message(self, message):
         try:

apps/hip-3-pusher/src/pusher/lazer_listener.py

@@ -3,8 +3,10 @@
 from loguru import logger
 import time
 import websockets
+from tenacity import retry, retry_if_exception_type, wait_exponential
 
-from pusher.config import Config
+from pusher.config import Config, STALE_TIMEOUT_SECONDS
+from pusher.exception import StaleConnection
 from pusher.price_state import PriceState, PriceUpdate
 
 
@@ -35,14 +37,13 @@ def get_subscribe_request(self, subscription_id: int):
     async def subscribe_all(self):
         await asyncio.gather(*(self.subscribe_single(router_url) for router_url in self.lazer_urls))
 
+    @retry(
+        retry=retry_if_exception_type((StaleConnection, websockets.ConnectionClosed)),
+        wait=wait_exponential(multiplier=1, min=1, max=10),
+        reraise=True,
+    )
     async def subscribe_single(self, router_url):
-        while True:
-            try:
-                await self.subscribe_single_inner(router_url)
-            except websockets.ConnectionClosed:
-                logger.error("Connection to {} closed; retrying", router_url)
-            except Exception as e:
-                logger.exception("Error on {}: {}", router_url, e)
+        return await self.subscribe_single_inner(router_url)
 
     async def subscribe_single_inner(self, router_url):
         headers = {
@@ -56,12 +57,19 @@ async def subscribe_single_inner(self, router_url):
             logger.info("Sent Lazer subscribe request to {}", router_url)
 
             # listen for updates
-            async for message in ws:
+            while True:
                 try:
+                    message = await asyncio.wait_for(ws.recv(), timeout=STALE_TIMEOUT_SECONDS)
                     data = json.loads(message)
                     self.parse_lazer_message(data)
+                except asyncio.TimeoutError:
+                    raise StaleConnection(f"No messages in {STALE_TIMEOUT_SECONDS} seconds, reconnecting")
+                except websockets.ConnectionClosed:
+                    raise
                 except json.JSONDecodeError as e:
                     logger.error("Failed to decode JSON message: {}", e)
+                except Exception as e:
+                    logger.error("Unexpected exception: {}", e)
 
     def parse_lazer_message(self, data):
         """