Merge pull request #61 from backend-developers-ltd/handle-miner-conne…

…ct-error

Handle miner connection refusal in validator

compute_horde/compute_horde/miner_client/base.py

@@ -2,6 +2,7 @@
 import asyncio
 import logging
 import random
+import time
 
 import websockets
 
@@ -10,10 +11,15 @@
 logger = logging.getLogger(__name__)
 
 
+class MinerConnectionError(Exception):
+    pass
+
+
 class AbstractMinerClient(abc.ABC):
 
     def __init__(self, loop: asyncio.AbstractEventLoop, miner_name: str):
         self.debounce_counter = 0
+        self.max_debounce_count: int | None = 5  # set to None for unlimited debounce
         self.loop = loop
         self.miner_name = miner_name
         self.ws: websockets.WebSocketClientProtocol | None = None
@@ -60,17 +66,27 @@ async def _connect(self):
         return await websockets.connect(self.miner_url())
 
     async def await_connect(self):
+        start_time = time.time()
         while True:
             try:
+                if self.max_debounce_count is not None and self.debounce_counter > self.max_debounce_count:
+                    time_took = time.time() - start_time
+                    raise MinerConnectionError(
+                        f'Could not connect to miner {self.miner_name} after {self.max_debounce_count} tries'
+                        f' in {time_took:0.2f} seconds'
+                    )
                 if self.debounce_counter:
                     sleep_time = self.sleep_time()
                     logger.info(f'Retrying connection to miner {self.miner_name} in {sleep_time:0.2f}')
                     await asyncio.sleep(sleep_time)
-                self.debounce_counter += 1
                 self.ws = await self._connect()
                 self.read_messages_task = self.loop.create_task(self.read_messages())
+                if self.debounce_counter:
+                    logger.info(f'Connected to miner {self.miner_name} after {self.debounce_counter + 1} attempts')
+                    self.debounce_counter = 0
                 return
             except (websockets.WebSocketException, OSError) as ex:
+                self.debounce_counter += 1
                 logger.info(f'Could not connect to miner {self.miner_name}: {str(ex)}')
 
     def sleep_time(self):

validator/app/src/compute_horde_validator/validator/management/commands/connect_facilitator.py

@@ -1,5 +1,6 @@
 import asyncio
 import base64
+import contextlib
 import io
 import logging
 import time
@@ -12,6 +13,7 @@
 import tenacity
 import websockets
 from asgiref.sync import async_to_sync, sync_to_async
+from compute_horde.miner_client.base import MinerConnectionError
 from compute_horde.mv_protocol.miner_requests import (
     V0DeclineJobRequest,
     V0ExecutorFailedRequest,
@@ -275,7 +277,22 @@ async def miner_driver(self, job_request: JobRequest):
             job_uuid=job_request.uuid,
             keypair=self.keypair,
         )
-        async with miner_client:
+        async with contextlib.AsyncExitStack() as exit_stack:
+            try:
+                await exit_stack.enter_async_context(miner_client)
+            except MinerConnectionError as exc:
+                comment = f'Miner connection error: {exc}'
+                logger.error(comment)
+                await self.send_model(JobStatusUpdate(
+                    uuid=job_request.uuid,
+                    status='failed',
+                    metadata=JobStatusMetadata(comment=comment),
+                ))
+                job.status = OrganicJob.Status.FAILED
+                job.comment = comment
+                await job.asave()
+                return
+
             job_timer = Timer(timeout=TOTAL_JOB_TIMEOUT)
             if job_request.input_url:
                 volume = Volume(volume_type=VolumeType.zip_url, contents=job_request.input_url)