[FIX] raise error if job does not terminate in tail_job_logs()

python/ray/dashboard/modules/job/sdk.py

@@ -482,8 +482,9 @@ async def tail_job_logs(self, job_id: str) -> AsyncIterator[str]:
             The iterator.
 
         Raises:
-            RuntimeError: If the job does not exist or if the request to the
-                job server fails.
+            RuntimeError: If the job does not exist, if the request to the
+                job server fails, or if the connection closes unexpectedly
+                before the job reaches a terminal state.
         """
         async with aiohttp.ClientSession(
             cookies=self._cookies, headers=self._headers
@@ -498,6 +499,15 @@ async def tail_job_logs(self, job_id: str) -> AsyncIterator[str]:
                 if msg.type == aiohttp.WSMsgType.TEXT:
                     yield msg.data
                 elif msg.type == aiohttp.WSMsgType.CLOSED:
+                    print(f"Close code: {ws.close_code}")
+                    if ws.close_code == aiohttp.WSCloseCode.ABNORMAL_CLOSURE:
+                        raise RuntimeError(
+                            f"WebSocket connection closed unexpectedly while job with close code {ws.close_code}"
+                        )
                     break
                 elif msg.type == aiohttp.WSMsgType.ERROR:
-                    pass
+                    # Old Ray versions may send ERROR on connection close
+                    raise RuntimeError(
+                        f"WebSocket error while tailing logs for job {job_id}. Err: {ws.exception()}"
+                    )
+                    break