trying to catch where TimeoutError and CancelledError become TypeError

due to trying to parse it as JSON when it should have been raised. aiohttp is weird with pause/unpause from a container.
runpod · Oct 11, 2024 · 58a573b · 58a573b
1 parent 303f45e
commit 58a573b
Show file tree

Hide file tree

Showing 2 changed files with 16 additions and 10 deletions.
diff --git a/runpod/serverless/modules/rp_job.py b/runpod/serverless/modules/rp_job.py
@@ -77,15 +77,18 @@ async def get_job(
             log.error(f"- Failed to get job, status code: {response.status}")
             return
 
-        try:
-            jobs = await response.json()
-            log.debug(f"- Job(s) Received")
-        except TypeError as err:
-            log.debug(f"- {response} | {err}")
-            raise response
-        except Exception as err:
-            log.debug(f"- {response} | {err}")
-            raise err
+        # Verify if the content type is JSON
+        if response.content_type != "application/json":
+            log.error(f"- Unexpected content type: {response.content_type}")
+            return
+
+        # Check if there is a non-empty content to parse
+        if response.content_length == 0:
+            log.debug("- No content to parse.")
+            return
+
+        jobs = await response.json()
+        log.debug(f"- Received Job(s)")
 
         # legacy job-take API
         if isinstance(jobs, dict):

diff --git a/runpod/serverless/modules/rp_scale.py b/runpod/serverless/modules/rp_scale.py
@@ -150,11 +150,14 @@ async def get_jobs(self, session: ClientSession):
                 acquired_jobs = await asyncio.wait_for(
                     get_job(session, jobs_needed), timeout=30
                 )
+            except asyncio.CancelledError:
+                log.debug("JobScaler.get_jobs | Request was cancelled.")
+                continue
             except TimeoutError:
                 log.debug("JobScaler.get_jobs | Job acquisition timed out. Retrying.")
                 continue
             except TypeError as error:
-                log.debug(f"Unexpected error: {error}. acquired_jobs={acquired_jobs}")
+                log.debug(f"JobScaler.get_jobs | Unexpected error: {error}.")
                 continue
             except Exception as error:
                 log.error(