From c19b827ccf9c60747d27dee69139979f9df16d0d Mon Sep 17 00:00:00 2001 From: Justin Date: Mon, 26 Oct 2026 14:10:01 -0400 Subject: [PATCH 01/13] fix(sentinel): poll /status when runsync returns before the job finishes runsync hands back IN_QUEUE or IN_PROGRESS when a job outlives the server-side wait, which is common on cold starts and model downloads. The sentinel returned that status dict as the function result, so callers failed later with a KeyError. Poll the sentinel /status route until the job reaches a terminal state, and raise on CANCELLED and TIMED_OUT instead of returning the status dict. --- src/runpod_flash/flash_sentinel.py | 55 +++++++++++- tests/unit/test_flash_sentinel.py | 132 +++++++++++++++++++++++++++++ 2 files changed, 186 insertions(+), 1 deletion(-) diff --git a/src/runpod_flash/flash_sentinel.py b/src/runpod_flash/flash_sentinel.py index e5a83219..4fbad38d 100644 --- a/src/runpod_flash/flash_sentinel.py +++ b/src/runpod_flash/flash_sentinel.py @@ -10,6 +10,7 @@ inspect.signature and sends the result as the runsync input body. """ +import asyncio import base64 import inspect import logging @@ -21,6 +22,7 @@ from .core.urls import ENDPOINT_DOMAIN from .core.utils import http as _http +from .core.utils.backoff import get_backoff_delay from .protos.remote_execution import FunctionRequest log = logging.getLogger(__name__) @@ -31,6 +33,13 @@ # FLASH_SENTINEL_TIMEOUT env var (seconds). DEFAULT_SENTINEL_TIMEOUT = 90 +# runsync returns one of these when the job outlives the server-side wait +# (cold starts, model downloads). the job keeps running, so poll /status. +_PENDING_STATUSES = frozenset({"IN_QUEUE", "IN_PROGRESS"}) + +# longest gap between /status polls, in seconds. +_STATUS_POLL_MAX_SECONDS = 5.0 + def _get_timeout(override: Optional[float] = None) -> float: """resolve the sentinel request timeout. @@ -108,7 +117,46 @@ async def _sentinel_qb_post( f"environment '{env}'. deploy it first with 'flash deploy'." ) response.raise_for_status() - return response.json() + data: Dict[str, Any] = response.json() + + if data.get("status") in _PENDING_STATUSES and data.get("id"): + data = await _poll_until_done( + app, env, endpoint_name, data["id"], effective_timeout + ) + return data + + +async def _poll_until_done( + app: str, + env: str, + endpoint_name: str, + job_id: str, + request_timeout: float, +) -> Dict[str, Any]: + """poll the sentinel /status route until the job leaves IN_QUEUE/IN_PROGRESS. + + there is no overall deadline here: the endpoint's execution timeout + ends the job server-side with TIMED_OUT, which this loop returns. + """ + url = f"{runpod.endpoint_url_base}/{FLASH_SENTINEL_ID}/status/{job_id}" + headers = _flash_headers(app, env, endpoint_name) + + log.debug("sentinel QB job %s still pending, polling /status", job_id) + + attempt = 0 + async with _http.get_authenticated_httpx_client(timeout=request_timeout) as client: + while True: + await asyncio.sleep( + get_backoff_delay( + attempt, base=1.0, max_seconds=_STATUS_POLL_MAX_SECONDS + ) + ) + attempt += 1 + response = await client.get(url, headers=headers) + response.raise_for_status() + data: Dict[str, Any] = response.json() + if data.get("status") not in _PENDING_STATUSES: + return data def _handle_sentinel_response(data: Dict[str, Any]) -> Any: @@ -122,6 +170,11 @@ def _handle_sentinel_response(data: Dict[str, Any]) -> Any: err = data.get("error") or data.get("output", {}).get("error", "unknown") raise RuntimeError(f"remote execution failed: {err}") + if data.get("status") in ("CANCELLED", "TIMED_OUT"): + raise RuntimeError( + f"remote execution {data['status'].lower()} (job {data.get('id', 'unknown')})" + ) + if "output" not in data and "status" not in data: raise RuntimeError( f"unexpected response from sentinel (no 'output' or 'status' key): {data}" diff --git a/tests/unit/test_flash_sentinel.py b/tests/unit/test_flash_sentinel.py index cfaf4051..9efeb9ed 100644 --- a/tests/unit/test_flash_sentinel.py +++ b/tests/unit/test_flash_sentinel.py @@ -154,6 +154,138 @@ async def my_func(): await sentinel_qb_execute("myapp", "prod", "gpu-worker", my_func) +class TestHandleSentinelTerminalStatuses: + @pytest.mark.parametrize("status", ["CANCELLED", "TIMED_OUT"]) + def test_cancelled_or_timed_out_raises(self, status): + from runpod_flash.flash_sentinel import _handle_sentinel_response + + data = {"id": "job-1", "status": status} + with pytest.raises(RuntimeError, match=status.lower()): + _handle_sentinel_response(data) + + +def _json_response(payload): + response = MagicMock() + response.status_code = 200 + response.raise_for_status = MagicMock() + response.json.return_value = payload + return response + + +class TestSentinelQBPolling: + """runsync returns IN_QUEUE/IN_PROGRESS when the job outlives the server wait.""" + + @pytest.fixture(autouse=True) + def no_sleep(self): + with patch("runpod_flash.flash_sentinel.asyncio.sleep", new=AsyncMock()): + yield + + @pytest.mark.asyncio + async def test_polls_status_until_completed(self, mock_httpx): + from runpod_flash.flash_sentinel import FLASH_SENTINEL_ID, sentinel_qb_execute + + mock_client = _make_mock_client( + _json_response({"id": "job-1", "status": "IN_QUEUE"}) + ) + mock_client.get = AsyncMock( + side_effect=[ + _json_response({"id": "job-1", "status": "IN_QUEUE"}), + _json_response({"id": "job-1", "status": "IN_PROGRESS"}), + _json_response( + {"id": "job-1", "status": "COMPLETED", "output": {"image": "b64"}} + ), + ] + ) + mock_httpx.return_value = mock_client + + async def my_func(prompt): + pass + + result = await sentinel_qb_execute( + "myapp", "prod", "gpu-worker", my_func, "a cat" + ) + + assert result == {"image": "b64"} + assert mock_client.get.call_count == 3 + status_call = mock_client.get.call_args + assert status_call.args[0].endswith(f"/{FLASH_SENTINEL_ID}/status/job-1") + assert status_call.kwargs["headers"]["X-Flash-Endpoint"] == "gpu-worker" + + @pytest.mark.asyncio + async def test_polled_failure_raises(self, mock_httpx): + from runpod_flash.flash_sentinel import sentinel_qb_execute + + mock_client = _make_mock_client( + _json_response({"id": "job-1", "status": "IN_PROGRESS"}) + ) + mock_client.get = AsyncMock( + return_value=_json_response( + {"id": "job-1", "status": "FAILED", "error": "CUDA out of memory"} + ) + ) + mock_httpx.return_value = mock_client + + async def my_func(): + pass + + with pytest.raises(RuntimeError, match="CUDA out of memory"): + await sentinel_qb_execute("myapp", "prod", "gpu-worker", my_func) + + @pytest.mark.asyncio + async def test_polled_timeout_raises(self, mock_httpx): + from runpod_flash.flash_sentinel import sentinel_qb_execute + + mock_client = _make_mock_client( + _json_response({"id": "job-1", "status": "IN_QUEUE"}) + ) + mock_client.get = AsyncMock( + return_value=_json_response({"id": "job-1", "status": "TIMED_OUT"}) + ) + mock_httpx.return_value = mock_client + + async def my_func(): + pass + + with pytest.raises(RuntimeError, match="timed_out"): + await sentinel_qb_execute("myapp", "prod", "gpu-worker", my_func) + + @pytest.mark.asyncio + async def test_class_execute_polls_too(self, mock_httpx): + from runpod_flash.flash_sentinel import sentinel_qb_class_execute + from runpod_flash.protos.remote_execution import FunctionRequest + + mock_client = _make_mock_client( + _json_response({"id": "job-2", "status": "IN_PROGRESS"}) + ) + mock_client.get = AsyncMock( + return_value=_json_response( + {"id": "job-2", "status": "COMPLETED", "output": {"ok": True}} + ) + ) + mock_httpx.return_value = mock_client + + request = FunctionRequest( + execution_type="class", class_name="MyModel", method_name="predict" + ) + result = await sentinel_qb_class_execute("myapp", "prod", "gpu-worker", request) + + assert result == {"ok": True} + mock_client.get.assert_awaited_once() + + @pytest.mark.asyncio + async def test_pending_without_id_is_not_polled(self, mock_httpx): + from runpod_flash.flash_sentinel import _sentinel_qb_post + + mock_client = _make_mock_client(_json_response({"status": "IN_QUEUE"})) + mock_client.get = AsyncMock() + mock_httpx.return_value = mock_client + + data = await _sentinel_qb_post("myapp", "prod", "gpu-worker", {"input": {}}) + + assert data == {"status": "IN_QUEUE"} + mock_client.get.assert_not_awaited() + + class TestSentinelQBClassExecute: @pytest.mark.asyncio async def test_dispatches_on_method_name(self, mock_httpx): From 2a448e03218516b54877085f7c0b31949134e5a7 Mon Sep 17 00:00:00 2001 From: Justin Date: Tue, 27 Oct 2026 10:15:45 -0400 Subject: [PATCH 02/13] fix(sentinel): ask runsync to return before the client times out runsync holds a request for 90s by default, the same as the sentinel's default client timeout, so on a cold start the client usually timed out before runsync could hand back IN_QUEUE/IN_PROGRESS. the /status polling never ran and callers saw a misleading 'endpoint may not be deployed' timeout. send wait= so runsync returns a pending status the client can poll. --- src/runpod_flash/flash_sentinel.py | 22 +++++++++++++++++++++- tests/unit/test_flash_sentinel.py | 28 ++++++++++++++++++++++++++++ 2 files changed, 49 insertions(+), 1 deletion(-) diff --git a/src/runpod_flash/flash_sentinel.py b/src/runpod_flash/flash_sentinel.py index 4fbad38d..c18dfc2b 100644 --- a/src/runpod_flash/flash_sentinel.py +++ b/src/runpod_flash/flash_sentinel.py @@ -40,6 +40,14 @@ # longest gap between /status polls, in seconds. _STATUS_POLL_MAX_SECONDS = 5.0 +# runsync holds the request for 90s by default, the same as the default +# client timeout, so the client could give up before runsync hands back a +# pending status. ask runsync to return this many seconds before the client +# timeout. ai-api accepts a wait of 1s to 300s. +_RUNSYNC_WAIT_MARGIN_SECONDS = 10.0 +_RUNSYNC_WAIT_MIN_SECONDS = 1.0 +_RUNSYNC_WAIT_MAX_SECONDS = 300.0 + def _get_timeout(override: Optional[float] = None) -> float: """resolve the sentinel request timeout. @@ -58,6 +66,13 @@ def _get_timeout(override: Optional[float] = None) -> float: return DEFAULT_SENTINEL_TIMEOUT +def _runsync_wait_ms(timeout: float) -> int: + """server-side runsync wait in ms, kept below the client timeout.""" + wait = timeout - _RUNSYNC_WAIT_MARGIN_SECONDS + wait = max(_RUNSYNC_WAIT_MIN_SECONDS, min(wait, _RUNSYNC_WAIT_MAX_SECONDS)) + return int(wait * 1000) + + def _flash_headers(app: str, env: str, endpoint: str) -> Dict[str, str]: """build the flash resolution headers.""" return { @@ -100,7 +115,12 @@ async def _sentinel_qb_post( async with _http.get_authenticated_httpx_client( timeout=effective_timeout ) as client: - response = await client.post(url, json=payload, headers=headers) + response = await client.post( + url, + json=payload, + headers=headers, + params={"wait": _runsync_wait_ms(effective_timeout)}, + ) except Exception as exc: if "timeout" in type(exc).__name__.lower() or "timeout" in str(exc).lower(): raise RuntimeError( diff --git a/tests/unit/test_flash_sentinel.py b/tests/unit/test_flash_sentinel.py index 9efeb9ed..9aef706f 100644 --- a/tests/unit/test_flash_sentinel.py +++ b/tests/unit/test_flash_sentinel.py @@ -286,6 +286,34 @@ async def test_pending_without_id_is_not_polled(self, mock_httpx): mock_client.get.assert_not_awaited() +class TestRunsyncWait: + """runsync must return a pending status before the client times out.""" + + @pytest.mark.parametrize( + ("timeout", "expected_ms"), + [(90, 80_000), (30, 20_000), (5, 1_000), (900, 300_000)], + ) + def test_wait_stays_below_client_timeout(self, timeout, expected_ms): + from runpod_flash.flash_sentinel import _runsync_wait_ms + + assert _runsync_wait_ms(timeout) == expected_ms + + @pytest.mark.asyncio + async def test_runsync_sends_wait_param(self, mock_httpx, monkeypatch): + from runpod_flash.flash_sentinel import _sentinel_qb_post + + monkeypatch.delenv("FLASH_SENTINEL_TIMEOUT", raising=False) + mock_client = _make_mock_client( + _json_response({"status": "COMPLETED", "output": {}}) + ) + mock_httpx.return_value = mock_client + + await _sentinel_qb_post("myapp", "prod", "gpu-worker", {"input": {"a": 1}}) + + assert mock_client.post.call_args.kwargs["params"] == {"wait": 80_000} + assert mock_httpx.call_args.kwargs["timeout"] == 90 + + class TestSentinelQBClassExecute: @pytest.mark.asyncio async def test_dispatches_on_method_name(self, mock_httpx): From 2f42032d44f7791032dee768c5cb9d309963d196 Mon Sep 17 00:00:00 2001 From: Justin Date: Tue, 27 Oct 2026 10:50:34 -0400 Subject: [PATCH 03/13] fix(deploy): print a curl example the endpoint can actually run the example sent {"input": {}}. runpod drops a job whose input is an empty dict, so the job never reached the worker and failed after about 40s with 'job timed out after 1 retries'. read the function's required parameters from source and use them as placeholders, or the __empty marker the handler already strips when the function takes none. --- src/runpod_flash/cli/commands/deploy.py | 65 ++++++++++++++++++-- tests/unit/cli/test_deploy.py | 80 +++++++++++++++++++++++++ 2 files changed, 140 insertions(+), 5 deletions(-) diff --git a/src/runpod_flash/cli/commands/deploy.py b/src/runpod_flash/cli/commands/deploy.py index 46a7fb61..e5a5d450 100644 --- a/src/runpod_flash/cli/commands/deploy.py +++ b/src/runpod_flash/cli/commands/deploy.py @@ -2,13 +2,17 @@ from __future__ import annotations +import ast import asyncio +import json import logging import shutil +from pathlib import Path from typing import Any import typer from rich.console import Console +from rich.markup import escape from runpod_flash.cli.utils.formatting import print_error from runpod_flash.core.exceptions import RunpodAPIKeyError @@ -116,15 +120,63 @@ def _handle_deploy_error(exc: Exception) -> None: raise typer.Exit(1) -def _print_curl_example(url: str, method: str = "POST") -> None: +def _required_params(file_path: str, func_name: str) -> list[str] | None: + """names of a function's parameters that have no default, read from source. + + returns None when the source can't be read or the function isn't found. + """ + try: + tree = ast.parse(Path(file_path).read_text()) + except (OSError, SyntaxError, ValueError): + return None + for node in ast.walk(tree): + if ( + isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) + and node.name == func_name + ): + args = node.args + positional = args.posonlyargs + args.args + required = positional[: len(positional) - len(args.defaults)] + required += [ + a + for a, default in zip(args.kwonlyargs, args.kw_defaults) + if default is None + ] + return [a.arg for a in required if a.arg not in ("self", "cls")] + return None + + +def _qb_example_input(resource: dict[str, Any]) -> dict[str, Any]: + """example runsync input for a queue-based endpoint. + + runpod drops a job whose input is an empty dict, so the job never + reaches the worker and times out. the deployed handler strips the + "__empty" marker, so use it when the function needs no arguments. + """ + functions = resource.get("functions") or [] + if not functions: + return {"__empty": True} + func = functions[0] + if func.get("is_class"): + return {"method": ""} + params = _required_params(resource.get("file_path", ""), func.get("name", "")) + if not params: + return {"__empty": True} + return {name: f"<{name}>" for name in params} + + +def _print_curl_example( + url: str, method: str = "POST", body: dict[str, Any] | None = None +) -> None: """Print a curl example for the given URL.""" lines = [f"curl -X {method} {url}"] if method == "POST": lines.append(' -H "Content-Type: application/json"') lines.append(' -H "Authorization: Bearer $RUNPOD_API_KEY"') if method == "POST": - lines.append(""" -d '{"input": {}}'""") - console.print("[dim]" + " \\\n".join(lines) + "[/dim]") + payload = json.dumps(body if body is not None else {"input": {}}) + lines.append(f" -d '{payload}'") + console.print("[dim]" + escape(" \\\n".join(lines)) + "[/dim]") def _display_post_deployment_guidance( @@ -187,8 +239,11 @@ def _display_post_deployment_guidance( # one curl example curl_url = None curl_method = "POST" + curl_body = None if qb_entries: - curl_url = f"{qb_entries[0][1]}/runsync" + qb_name, qb_url = qb_entries[0] + curl_url = f"{qb_url}/runsync" + curl_body = {"input": _qb_example_input(resources.get(qb_name, {}))} elif lb_entries: _name, base_url, lb_routes = lb_entries[0] for k in sorted(lb_routes.keys()): @@ -205,7 +260,7 @@ def _display_post_deployment_guidance( if curl_url: console.print() - _print_curl_example(curl_url, method=curl_method) + _print_curl_example(curl_url, method=curl_method, body=curl_body) def _launch_preview(project_dir): diff --git a/tests/unit/cli/test_deploy.py b/tests/unit/cli/test_deploy.py index ae046b1a..7aa9f4e8 100644 --- a/tests/unit/cli/test_deploy.py +++ b/tests/unit/cli/test_deploy.py @@ -801,6 +801,86 @@ def test_qb_curl_uses_runsync(self, patched_console): assert "curl" in output assert "curl -X POST https://def.api.runpod.ai/runsync" in output + def test_qb_curl_sends_required_params(self, patched_console, tmp_path): + from runpod_flash.cli.commands.deploy import _display_post_deployment_guidance + + src = tmp_path / "gpu_demo.py" + src.write_text( + "async def gpu_matrix_multiply(size, *, scale, seed=0, verbose=False):\n" + " return size\n" + ) + _display_post_deployment_guidance( + env_name="production", + resources_endpoints={"my_qb": "https://api.runpod.ai/v2/abc123"}, + resources={ + "my_qb": { + "file_path": str(src), + "functions": [{"name": "gpu_matrix_multiply", "is_class": False}], + } + }, + routes={}, + ) + output = self._collect_output(patched_console) + assert '{"input": {"size": "", "scale": ""}}' in output + assert '{"input": {}}' not in output + + def test_qb_curl_marks_empty_input_for_no_arg_function( + self, patched_console, tmp_path + ): + from runpod_flash.cli.commands.deploy import _display_post_deployment_guidance + + src = tmp_path / "jobs.py" + src.write_text("def ping(verbose=False):\n return 'pong'\n") + _display_post_deployment_guidance( + env_name="production", + resources_endpoints={"my_qb": "https://api.runpod.ai/v2/abc123"}, + resources={ + "my_qb": { + "file_path": str(src), + "functions": [{"name": "ping", "is_class": False}], + } + }, + routes={}, + ) + output = self._collect_output(patched_console) + assert '{"input": {"__empty": true}}' in output + + def test_qb_curl_never_sends_empty_input(self, patched_console): + """runpod drops jobs with an empty input dict, so the example must not.""" + from runpod_flash.cli.commands.deploy import _display_post_deployment_guidance + + _display_post_deployment_guidance( + env_name="production", + resources_endpoints={"my_qb": "https://api.runpod.ai/v2/abc123"}, + resources={ + "my_qb": { + "file_path": "missing.py", + "functions": [{"name": "f", "is_class": False}], + } + }, + routes={}, + ) + output = self._collect_output(patched_console) + assert '{"input": {"__empty": true}}' in output + assert '{"input": {}}' not in output + + def test_qb_curl_for_class_endpoint_names_method(self, patched_console): + from runpod_flash.cli.commands.deploy import _display_post_deployment_guidance + + _display_post_deployment_guidance( + env_name="production", + resources_endpoints={"my_qb": "https://api.runpod.ai/v2/abc123"}, + resources={ + "my_qb": { + "file_path": "model.py", + "functions": [{"name": "Model", "is_class": True}], + } + }, + routes={}, + ) + output = self._collect_output(patched_console) + assert '{"input": {"method": ""}}' in output + def test_lb_endpoint_shows_routes(self, patched_console): from runpod_flash.cli.commands.deploy import _display_post_deployment_guidance From 8a09c8ba2fc220a5fd86494c28b47175e3d5958f Mon Sep 17 00:00:00 2001 From: Justin Date: Tue, 27 Oct 2026 11:01:05 -0400 Subject: [PATCH 04/13] fix(sentinel): harden polling and the curl example after review - raise when runsync returns a pending status without a job id - retry transient network errors while polling /status (5 in a row max) - cancel and raise when a job stays IN_QUEUE for 15 minutes, instead of polling forever when no worker can start - curl example: use a real method name for multi-method class endpoints, omit 'method' for single-method classes, read only top-level definitions --- src/runpod_flash/cli/commands/deploy.py | 43 ++++++- src/runpod_flash/flash_sentinel.py | 62 +++++++++- tests/unit/cli/test_deploy.py | 63 +++++++++- tests/unit/test_flash_sentinel.py | 145 +++++++++++++++++++++++- 4 files changed, 293 insertions(+), 20 deletions(-) diff --git a/src/runpod_flash/cli/commands/deploy.py b/src/runpod_flash/cli/commands/deploy.py index e5a5d450..f2e4e7e3 100644 --- a/src/runpod_flash/cli/commands/deploy.py +++ b/src/runpod_flash/cli/commands/deploy.py @@ -120,16 +120,35 @@ def _handle_deploy_error(exc: Exception) -> None: raise typer.Exit(1) -def _required_params(file_path: str, func_name: str) -> list[str] | None: +def _required_params( + file_path: str, func_name: str, class_name: str | None = None +) -> list[str] | None: """names of a function's parameters that have no default, read from source. - returns None when the source can't be read or the function isn't found. + looks only at top-level definitions (or methods of a top-level class), + which is what the generated handler imports. returns None when the + source can't be read or the definition isn't found. """ try: tree = ast.parse(Path(file_path).read_text()) except (OSError, SyntaxError, ValueError): return None - for node in ast.walk(tree): + + body: list[ast.stmt] = tree.body + if class_name is not None: + cls = next( + ( + n + for n in tree.body + if isinstance(n, ast.ClassDef) and n.name == class_name + ), + None, + ) + if cls is None: + return None + body = cls.body + + for node in body: if ( isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) and node.name == func_name @@ -151,15 +170,27 @@ def _qb_example_input(resource: dict[str, Any]) -> dict[str, Any]: runpod drops a job whose input is an empty dict, so the job never reaches the worker and times out. the deployed handler strips the - "__empty" marker, so use it when the function needs no arguments. + "__empty" marker, so use it when the call needs no arguments. """ functions = resource.get("functions") or [] if not functions: return {"__empty": True} func = functions[0] + file_path = resource.get("file_path", "") + if func.get("is_class"): - return {"method": ""} - params = _required_params(resource.get("file_path", ""), func.get("name", "")) + methods = func.get("class_methods") or [] + if not methods: + return {"method": ""} + method = methods[0] + params = _required_params(file_path, method, class_name=func.get("name")) + body: dict[str, Any] = {name: f"<{name}>" for name in params or []} + # class handlers dispatch on "method" only when there's more than one + if len(methods) > 1: + body = {"method": method, **body} + return body or {"__empty": True} + + params = _required_params(file_path, func.get("name", "")) if not params: return {"__empty": True} return {name: f"<{name}>" for name in params} diff --git a/src/runpod_flash/flash_sentinel.py b/src/runpod_flash/flash_sentinel.py index c18dfc2b..fa600316 100644 --- a/src/runpod_flash/flash_sentinel.py +++ b/src/runpod_flash/flash_sentinel.py @@ -15,9 +15,11 @@ import inspect import logging import os +import time from typing import Any, Callable, Dict, Optional import cloudpickle +import httpx import runpod from .core.urls import ENDPOINT_DOMAIN @@ -40,6 +42,13 @@ # longest gap between /status polls, in seconds. _STATUS_POLL_MAX_SECONDS = 5.0 +# give up (and cancel) when no worker picks the job up for this long. +# once the job is IN_PROGRESS, the endpoint's execution timeout bounds it. +_QUEUE_WAIT_MAX_SECONDS = 900.0 + +# consecutive network errors tolerated while polling /status. +_STATUS_POLL_MAX_TRANSPORT_ERRORS = 5 + # runsync holds the request for 90s by default, the same as the default # client timeout, so the client could give up before runsync hands back a # pending status. ask runsync to return this many seconds before the client @@ -139,7 +148,12 @@ async def _sentinel_qb_post( response.raise_for_status() data: Dict[str, Any] = response.json() - if data.get("status") in _PENDING_STATUSES and data.get("id"): + if data.get("status") in _PENDING_STATUSES: + if not data.get("id"): + raise RuntimeError( + f"endpoint '{endpoint_name}' returned status " + f"{data['status']} without a job id: {data}" + ) data = await _poll_until_done( app, env, endpoint_name, data["id"], effective_timeout ) @@ -155,15 +169,18 @@ async def _poll_until_done( ) -> Dict[str, Any]: """poll the sentinel /status route until the job leaves IN_QUEUE/IN_PROGRESS. - there is no overall deadline here: the endpoint's execution timeout - ends the job server-side with TIMED_OUT, which this loop returns. + a running job is bounded by the endpoint's execution timeout, which ends + it server-side with TIMED_OUT. a job no worker picks up is cancelled + after _QUEUE_WAIT_MAX_SECONDS. """ - url = f"{runpod.endpoint_url_base}/{FLASH_SENTINEL_ID}/status/{job_id}" + base = f"{runpod.endpoint_url_base}/{FLASH_SENTINEL_ID}" headers = _flash_headers(app, env, endpoint_name) log.debug("sentinel QB job %s still pending, polling /status", job_id) + queued_since = time.monotonic() attempt = 0 + transport_errors = 0 async with _http.get_authenticated_httpx_client(timeout=request_timeout) as client: while True: await asyncio.sleep( @@ -172,11 +189,44 @@ async def _poll_until_done( ) ) attempt += 1 - response = await client.get(url, headers=headers) + try: + response = await client.get(f"{base}/status/{job_id}", headers=headers) + except httpx.TransportError as exc: + transport_errors += 1 + if transport_errors >= _STATUS_POLL_MAX_TRANSPORT_ERRORS: + raise RuntimeError( + f"lost contact with endpoint '{endpoint_name}' while " + f"waiting for job {job_id}: {exc}" + ) from exc + log.debug("polling job %s failed (%s), retrying", job_id, exc) + continue + transport_errors = 0 response.raise_for_status() data: Dict[str, Any] = response.json() - if data.get("status") not in _PENDING_STATUSES: + status = data.get("status") + if status not in _PENDING_STATUSES: return data + if status != "IN_QUEUE": + queued_since = time.monotonic() + elif time.monotonic() - queued_since > _QUEUE_WAIT_MAX_SECONDS: + await _cancel_job(client, base, headers, job_id) + raise RuntimeError( + f"job {job_id} on endpoint '{endpoint_name}' was still " + f"queued after {_QUEUE_WAIT_MAX_SECONDS:.0f}s, so it was " + f"cancelled. no worker picked it up; check that GPUs of " + f"the requested type are available." + ) + + +async def _cancel_job( + client: httpx.AsyncClient, base: str, headers: Dict[str, str], job_id: str +) -> None: + """best-effort cancel so an abandoned job doesn't run later.""" + try: + response = await client.post(f"{base}/cancel/{job_id}", headers=headers) + response.raise_for_status() + except Exception as exc: + log.warning("could not cancel job %s: %s", job_id, exc) def _handle_sentinel_response(data: Dict[str, Any]) -> Any: diff --git a/tests/unit/cli/test_deploy.py b/tests/unit/cli/test_deploy.py index 7aa9f4e8..94f0ac34 100644 --- a/tests/unit/cli/test_deploy.py +++ b/tests/unit/cli/test_deploy.py @@ -864,22 +864,77 @@ def test_qb_curl_never_sends_empty_input(self, patched_console): assert '{"input": {"__empty": true}}' in output assert '{"input": {}}' not in output - def test_qb_curl_for_class_endpoint_names_method(self, patched_console): + def _class_resource(self, src, methods): + return { + "my_qb": { + "file_path": str(src), + "functions": [ + {"name": "Model", "is_class": True, "class_methods": methods} + ], + } + } + + def test_qb_curl_for_single_method_class_omits_method( + self, patched_console, tmp_path + ): from runpod_flash.cli.commands.deploy import _display_post_deployment_guidance + src = tmp_path / "model.py" + src.write_text( + "class Model:\n def predict(self, image, k=1):\n ...\n" + ) + _display_post_deployment_guidance( + env_name="production", + resources_endpoints={"my_qb": "https://api.runpod.ai/v2/abc123"}, + resources=self._class_resource(src, ["predict"]), + routes={}, + ) + output = self._collect_output(patched_console) + assert '{"input": {"image": ""}}' in output + + def test_qb_curl_for_multi_method_class_names_real_method( + self, patched_console, tmp_path + ): + from runpod_flash.cli.commands.deploy import _display_post_deployment_guidance + + src = tmp_path / "model.py" + src.write_text( + "class Model:\n" + " def predict(self, image):\n ...\n" + " def embed(self, text):\n ...\n" + ) + _display_post_deployment_guidance( + env_name="production", + resources_endpoints={"my_qb": "https://api.runpod.ai/v2/abc123"}, + resources=self._class_resource(src, ["predict", "embed"]), + routes={}, + ) + output = self._collect_output(patched_console) + assert '{"input": {"method": "predict", "image": ""}}' in output + + def test_qb_curl_reads_top_level_function_not_nested_namesake( + self, patched_console, tmp_path + ): + from runpod_flash.cli.commands.deploy import _display_post_deployment_guidance + + src = tmp_path / "jobs.py" + src.write_text( + "class Helper:\n def predict(self, image):\n ...\n\n" + "def predict(prompt):\n ...\n" + ) _display_post_deployment_guidance( env_name="production", resources_endpoints={"my_qb": "https://api.runpod.ai/v2/abc123"}, resources={ "my_qb": { - "file_path": "model.py", - "functions": [{"name": "Model", "is_class": True}], + "file_path": str(src), + "functions": [{"name": "predict", "is_class": False}], } }, routes={}, ) output = self._collect_output(patched_console) - assert '{"input": {"method": ""}}' in output + assert '{"input": {"prompt": ""}}' in output def test_lb_endpoint_shows_routes(self, patched_console): from runpod_flash.cli.commands.deploy import _display_post_deployment_guidance diff --git a/tests/unit/test_flash_sentinel.py b/tests/unit/test_flash_sentinel.py index 9aef706f..e61cde6b 100644 --- a/tests/unit/test_flash_sentinel.py +++ b/tests/unit/test_flash_sentinel.py @@ -273,18 +273,155 @@ async def test_class_execute_polls_too(self, mock_httpx): mock_client.get.assert_awaited_once() @pytest.mark.asyncio - async def test_pending_without_id_is_not_polled(self, mock_httpx): + async def test_pending_without_id_raises(self, mock_httpx): from runpod_flash.flash_sentinel import _sentinel_qb_post mock_client = _make_mock_client(_json_response({"status": "IN_QUEUE"})) mock_client.get = AsyncMock() mock_httpx.return_value = mock_client - data = await _sentinel_qb_post("myapp", "prod", "gpu-worker", {"input": {}}) - - assert data == {"status": "IN_QUEUE"} + with pytest.raises(RuntimeError, match="without a job id"): + await _sentinel_qb_post("myapp", "prod", "gpu-worker", {"input": {}}) mock_client.get.assert_not_awaited() + @pytest.mark.asyncio + async def test_transient_poll_error_is_retried(self, mock_httpx): + import httpx + + from runpod_flash.flash_sentinel import sentinel_qb_execute + + mock_client = _make_mock_client( + _json_response({"id": "job-1", "status": "IN_PROGRESS"}) + ) + mock_client.get = AsyncMock( + side_effect=[ + httpx.ReadTimeout("slow"), + _json_response( + {"id": "job-1", "status": "COMPLETED", "output": {"ok": True}} + ), + ] + ) + mock_httpx.return_value = mock_client + + async def my_func(): + pass + + result = await sentinel_qb_execute("myapp", "prod", "gpu-worker", my_func) + + assert result == {"ok": True} + assert mock_client.get.call_count == 2 + + @pytest.mark.asyncio + async def test_repeated_poll_errors_raise(self, mock_httpx): + import httpx + + from runpod_flash.flash_sentinel import ( + _STATUS_POLL_MAX_TRANSPORT_ERRORS, + sentinel_qb_execute, + ) + + mock_client = _make_mock_client( + _json_response({"id": "job-1", "status": "IN_PROGRESS"}) + ) + mock_client.get = AsyncMock(side_effect=httpx.ConnectError("down")) + mock_httpx.return_value = mock_client + + async def my_func(): + pass + + with pytest.raises(RuntimeError, match="lost contact.*job-1"): + await sentinel_qb_execute("myapp", "prod", "gpu-worker", my_func) + assert mock_client.get.call_count == _STATUS_POLL_MAX_TRANSPORT_ERRORS + + @pytest.mark.asyncio + async def test_poll_http_error_propagates(self, mock_httpx): + import httpx + + from runpod_flash.flash_sentinel import sentinel_qb_execute + + error_response = MagicMock() + error_response.raise_for_status = MagicMock( + side_effect=httpx.HTTPStatusError( + "500", request=MagicMock(), response=MagicMock() + ) + ) + mock_client = _make_mock_client( + _json_response({"id": "job-1", "status": "IN_PROGRESS"}) + ) + mock_client.get = AsyncMock(return_value=error_response) + mock_httpx.return_value = mock_client + + async def my_func(): + pass + + with pytest.raises(httpx.HTTPStatusError): + await sentinel_qb_execute("myapp", "prod", "gpu-worker", my_func) + mock_client.get.assert_awaited_once() + + @pytest.mark.asyncio + async def test_job_stuck_in_queue_is_cancelled(self, mock_httpx): + from runpod_flash.flash_sentinel import ( + _QUEUE_WAIT_MAX_SECONDS, + FLASH_SENTINEL_ID, + sentinel_qb_execute, + ) + + queued = _json_response({"id": "job-1", "status": "IN_QUEUE"}) + mock_client = _make_mock_client(queued) + mock_client.get = AsyncMock(return_value=queued) + mock_httpx.return_value = mock_client + + clock = iter([0.0, 10.0, _QUEUE_WAIT_MAX_SECONDS + 1]) + + async def my_func(): + pass + + with patch( + "runpod_flash.flash_sentinel.time.monotonic", + side_effect=lambda: next(clock), + ): + with pytest.raises(RuntimeError, match="still queued.*cancelled"): + await sentinel_qb_execute("myapp", "prod", "gpu-worker", my_func) + + cancel_url = mock_client.post.call_args.args[0] + assert cancel_url.endswith(f"/{FLASH_SENTINEL_ID}/cancel/job-1") + + @pytest.mark.asyncio + async def test_running_job_is_not_cancelled_by_queue_deadline(self, mock_httpx): + from runpod_flash.flash_sentinel import ( + _QUEUE_WAIT_MAX_SECONDS, + sentinel_qb_execute, + ) + + mock_client = _make_mock_client( + _json_response({"id": "job-1", "status": "IN_QUEUE"}) + ) + mock_client.get = AsyncMock( + side_effect=[ + _json_response({"id": "job-1", "status": "IN_PROGRESS"}), + _json_response({"id": "job-1", "status": "IN_PROGRESS"}), + _json_response( + {"id": "job-1", "status": "COMPLETED", "output": {"ok": 1}} + ), + ] + ) + mock_httpx.return_value = mock_client + + far = _QUEUE_WAIT_MAX_SECONDS * 10 + clock = iter([0.0, far, far * 2, far * 3]) + + async def my_func(): + pass + + with patch( + "runpod_flash.flash_sentinel.time.monotonic", + side_effect=lambda: next(clock), + ): + result = await sentinel_qb_execute("myapp", "prod", "gpu-worker", my_func) + + assert result == {"ok": 1} + assert mock_client.post.await_count == 1 # the runsync call, no cancel + class TestRunsyncWait: """runsync must return a pending status before the client times out.""" From 5d467d493cd1e114a6d34ddcf83756df33f9db30 Mon Sep 17 00:00:00 2001 From: Justin Date: Tue, 27 Oct 2026 11:04:59 -0400 Subject: [PATCH 05/13] fix(sentinel): retry transient status errors and report failed cancels - treat 429 and 5xx from /status like network errors (bounded retries) - say so when cancelling a stuck job fails instead of claiming it was cancelled - curl example: drop self/cls only as a class method's receiver --- src/runpod_flash/cli/commands/deploy.py | 9 ++- src/runpod_flash/flash_sentinel.py | 23 +++++--- tests/unit/cli/test_deploy.py | 21 +++++++ tests/unit/test_flash_sentinel.py | 73 ++++++++++++++++++++++--- 4 files changed, 111 insertions(+), 15 deletions(-) diff --git a/src/runpod_flash/cli/commands/deploy.py b/src/runpod_flash/cli/commands/deploy.py index f2e4e7e3..1faee74c 100644 --- a/src/runpod_flash/cli/commands/deploy.py +++ b/src/runpod_flash/cli/commands/deploy.py @@ -161,7 +161,14 @@ def _required_params( for a, default in zip(args.kwonlyargs, args.kw_defaults) if default is None ] - return [a.arg for a in required if a.arg not in ("self", "cls")] + # a method's receiver is bound by the handler, not sent as input + if class_name is not None and required and positional: + if required[0] is positional[0] and positional[0].arg in ( + "self", + "cls", + ): + required = required[1:] + return [a.arg for a in required] return None diff --git a/src/runpod_flash/flash_sentinel.py b/src/runpod_flash/flash_sentinel.py index fa600316..154b89d5 100644 --- a/src/runpod_flash/flash_sentinel.py +++ b/src/runpod_flash/flash_sentinel.py @@ -46,7 +46,7 @@ # once the job is IN_PROGRESS, the endpoint's execution timeout bounds it. _QUEUE_WAIT_MAX_SECONDS = 900.0 -# consecutive network errors tolerated while polling /status. +# consecutive network errors or 429/5xx responses tolerated while polling. _STATUS_POLL_MAX_TRANSPORT_ERRORS = 5 # runsync holds the request for 90s by default, the same as the default @@ -191,7 +191,9 @@ async def _poll_until_done( attempt += 1 try: response = await client.get(f"{base}/status/{job_id}", headers=headers) - except httpx.TransportError as exc: + if response.status_code == 429 or response.status_code >= 500: + response.raise_for_status() + except (httpx.TransportError, httpx.HTTPStatusError) as exc: transport_errors += 1 if transport_errors >= _STATUS_POLL_MAX_TRANSPORT_ERRORS: raise RuntimeError( @@ -209,24 +211,31 @@ async def _poll_until_done( if status != "IN_QUEUE": queued_since = time.monotonic() elif time.monotonic() - queued_since > _QUEUE_WAIT_MAX_SECONDS: - await _cancel_job(client, base, headers, job_id) + cancelled = await _cancel_job(client, base, headers, job_id) + outcome = ( + "it was cancelled" + if cancelled + else "cancelling it failed, so it may still run later" + ) raise RuntimeError( f"job {job_id} on endpoint '{endpoint_name}' was still " - f"queued after {_QUEUE_WAIT_MAX_SECONDS:.0f}s, so it was " - f"cancelled. no worker picked it up; check that GPUs of " - f"the requested type are available." + f"queued after {_QUEUE_WAIT_MAX_SECONDS:.0f}s and {outcome}. " + f"no worker picked it up; check that GPUs of the requested " + f"type are available." ) async def _cancel_job( client: httpx.AsyncClient, base: str, headers: Dict[str, str], job_id: str -) -> None: +) -> bool: """best-effort cancel so an abandoned job doesn't run later.""" try: response = await client.post(f"{base}/cancel/{job_id}", headers=headers) response.raise_for_status() except Exception as exc: log.warning("could not cancel job %s: %s", job_id, exc) + return False + return True def _handle_sentinel_response(data: Dict[str, Any]) -> Any: diff --git a/tests/unit/cli/test_deploy.py b/tests/unit/cli/test_deploy.py index 94f0ac34..3b27d779 100644 --- a/tests/unit/cli/test_deploy.py +++ b/tests/unit/cli/test_deploy.py @@ -936,6 +936,27 @@ def test_qb_curl_reads_top_level_function_not_nested_namesake( output = self._collect_output(patched_console) assert '{"input": {"prompt": ""}}' in output + def test_qb_curl_keeps_self_param_on_plain_function( + self, patched_console, tmp_path + ): + from runpod_flash.cli.commands.deploy import _display_post_deployment_guidance + + src = tmp_path / "jobs.py" + src.write_text("def run(self):\n ...\n") + _display_post_deployment_guidance( + env_name="production", + resources_endpoints={"my_qb": "https://api.runpod.ai/v2/abc123"}, + resources={ + "my_qb": { + "file_path": str(src), + "functions": [{"name": "run", "is_class": False}], + } + }, + routes={}, + ) + output = self._collect_output(patched_console) + assert '{"input": {"self": ""}}' in output + def test_lb_endpoint_shows_routes(self, patched_console): from runpod_flash.cli.commands.deploy import _display_post_deployment_guidance diff --git a/tests/unit/test_flash_sentinel.py b/tests/unit/test_flash_sentinel.py index e61cde6b..e7baebe9 100644 --- a/tests/unit/test_flash_sentinel.py +++ b/tests/unit/test_flash_sentinel.py @@ -172,6 +172,19 @@ def _json_response(payload): return response +def _http_error_response(status_code): + import httpx + + response = MagicMock() + response.status_code = status_code + response.raise_for_status = MagicMock( + side_effect=httpx.HTTPStatusError( + str(status_code), request=MagicMock(), response=MagicMock() + ) + ) + return response + + class TestSentinelQBPolling: """runsync returns IN_QUEUE/IN_PROGRESS when the job outlives the server wait.""" @@ -334,17 +347,12 @@ async def my_func(): assert mock_client.get.call_count == _STATUS_POLL_MAX_TRANSPORT_ERRORS @pytest.mark.asyncio - async def test_poll_http_error_propagates(self, mock_httpx): + async def test_poll_client_error_propagates(self, mock_httpx): import httpx from runpod_flash.flash_sentinel import sentinel_qb_execute - error_response = MagicMock() - error_response.raise_for_status = MagicMock( - side_effect=httpx.HTTPStatusError( - "500", request=MagicMock(), response=MagicMock() - ) - ) + error_response = _http_error_response(404) mock_client = _make_mock_client( _json_response({"id": "job-1", "status": "IN_PROGRESS"}) ) @@ -358,6 +366,32 @@ async def my_func(): await sentinel_qb_execute("myapp", "prod", "gpu-worker", my_func) mock_client.get.assert_awaited_once() + @pytest.mark.asyncio + @pytest.mark.parametrize("status_code", [429, 503]) + async def test_poll_recovers_from_transient_status(self, mock_httpx, status_code): + from runpod_flash.flash_sentinel import sentinel_qb_execute + + mock_client = _make_mock_client( + _json_response({"id": "job-1", "status": "IN_PROGRESS"}) + ) + mock_client.get = AsyncMock( + side_effect=[ + _http_error_response(status_code), + _json_response( + {"id": "job-1", "status": "COMPLETED", "output": {"ok": True}} + ), + ] + ) + mock_httpx.return_value = mock_client + + async def my_func(): + pass + + result = await sentinel_qb_execute("myapp", "prod", "gpu-worker", my_func) + + assert result == {"ok": True} + assert mock_client.get.call_count == 2 + @pytest.mark.asyncio async def test_job_stuck_in_queue_is_cancelled(self, mock_httpx): from runpod_flash.flash_sentinel import ( @@ -386,6 +420,31 @@ async def my_func(): cancel_url = mock_client.post.call_args.args[0] assert cancel_url.endswith(f"/{FLASH_SENTINEL_ID}/cancel/job-1") + @pytest.mark.asyncio + async def test_failed_cancel_is_reported(self, mock_httpx): + from runpod_flash.flash_sentinel import ( + _QUEUE_WAIT_MAX_SECONDS, + sentinel_qb_execute, + ) + + queued = _json_response({"id": "job-1", "status": "IN_QUEUE"}) + mock_client = _make_mock_client(queued) + mock_client.post = AsyncMock(side_effect=[queued, _http_error_response(500)]) + mock_client.get = AsyncMock(return_value=queued) + mock_httpx.return_value = mock_client + + clock = iter([0.0, _QUEUE_WAIT_MAX_SECONDS + 1]) + + async def my_func(): + pass + + with patch( + "runpod_flash.flash_sentinel.time.monotonic", + side_effect=lambda: next(clock), + ): + with pytest.raises(RuntimeError, match="cancelling it failed"): + await sentinel_qb_execute("myapp", "prod", "gpu-worker", my_func) + @pytest.mark.asyncio async def test_running_job_is_not_cancelled_by_queue_deadline(self, mock_httpx): from runpod_flash.flash_sentinel import ( From 00577dc2479871b4529251a0af8c68ac30c74059 Mon Sep 17 00:00:00 2001 From: Justin Date: Tue, 27 Oct 2026 11:08:33 -0400 Subject: [PATCH 06/13] fix(sentinel): submit with /run when the timeout leaves no room for runsync runsync waits at least 1s server-side, so a client timeout at or below the 10s margin plus that minimum could expire before runsync returned a job id. submit with /run in that case and poll. --- src/runpod_flash/flash_sentinel.py | 17 +++++++++-- tests/unit/test_flash_sentinel.py | 48 ++++++++++++++++++++++++++++++ 2 files changed, 62 insertions(+), 3 deletions(-) diff --git a/src/runpod_flash/flash_sentinel.py b/src/runpod_flash/flash_sentinel.py index 154b89d5..89a76b93 100644 --- a/src/runpod_flash/flash_sentinel.py +++ b/src/runpod_flash/flash_sentinel.py @@ -113,10 +113,21 @@ async def _sentinel_qb_post( payload: Dict[str, Any], timeout: Optional[float] = None, ) -> Dict[str, Any]: - """post a payload to the sentinel runsync URL and return the raw response dict.""" - url = f"{runpod.endpoint_url_base}/{FLASH_SENTINEL_ID}/runsync" + """submit a payload through the sentinel and return the final response dict. + + uses /runsync with a server-side wait below the client timeout. a + timeout too short to leave room for runsync's minimum wait submits + with /run instead. either way, a pending job is polled to completion. + """ headers = _flash_headers(app, env, endpoint_name) effective_timeout = _get_timeout(timeout) + base = f"{runpod.endpoint_url_base}/{FLASH_SENTINEL_ID}" + if effective_timeout > _RUNSYNC_WAIT_MARGIN_SECONDS + _RUNSYNC_WAIT_MIN_SECONDS: + url = f"{base}/runsync" + params: Dict[str, Any] = {"wait": _runsync_wait_ms(effective_timeout)} + else: + url = f"{base}/run" + params = {} log.debug("sentinel QB -> %s/%s/%s", app, env, endpoint_name) @@ -128,7 +139,7 @@ async def _sentinel_qb_post( url, json=payload, headers=headers, - params={"wait": _runsync_wait_ms(effective_timeout)}, + params=params, ) except Exception as exc: if "timeout" in type(exc).__name__.lower() or "timeout" in str(exc).lower(): diff --git a/tests/unit/test_flash_sentinel.py b/tests/unit/test_flash_sentinel.py index e7baebe9..01c46803 100644 --- a/tests/unit/test_flash_sentinel.py +++ b/tests/unit/test_flash_sentinel.py @@ -510,6 +510,54 @@ async def test_runsync_sends_wait_param(self, mock_httpx, monkeypatch): assert mock_httpx.call_args.kwargs["timeout"] == 90 +class TestShortTimeoutUsesRun: + @pytest.fixture(autouse=True) + def no_sleep(self): + with patch("runpod_flash.flash_sentinel.asyncio.sleep", new=AsyncMock()): + yield + + @pytest.mark.asyncio + @pytest.mark.parametrize("timeout", [1, 11]) + async def test_short_timeout_submits_with_run(self, mock_httpx, timeout): + from runpod_flash.flash_sentinel import FLASH_SENTINEL_ID, _sentinel_qb_post + + mock_client = _make_mock_client( + _json_response({"id": "job-1", "status": "IN_QUEUE"}) + ) + mock_client.get = AsyncMock( + return_value=_json_response( + {"id": "job-1", "status": "COMPLETED", "output": {"ok": True}} + ) + ) + mock_httpx.return_value = mock_client + + data = await _sentinel_qb_post( + "myapp", "prod", "gpu-worker", {"input": {"a": 1}}, timeout=timeout + ) + + assert data["output"] == {"ok": True} + call = mock_client.post.call_args + assert call.args[0].endswith(f"/{FLASH_SENTINEL_ID}/run") + assert call.kwargs["params"] == {} + + @pytest.mark.asyncio + async def test_timeout_above_margin_uses_runsync(self, mock_httpx): + from runpod_flash.flash_sentinel import FLASH_SENTINEL_ID, _sentinel_qb_post + + mock_client = _make_mock_client( + _json_response({"status": "COMPLETED", "output": {}}) + ) + mock_httpx.return_value = mock_client + + await _sentinel_qb_post( + "myapp", "prod", "gpu-worker", {"input": {"a": 1}}, timeout=12 + ) + + call = mock_client.post.call_args + assert call.args[0].endswith(f"/{FLASH_SENTINEL_ID}/runsync") + assert call.kwargs["params"] == {"wait": 2_000} + + class TestSentinelQBClassExecute: @pytest.mark.asyncio async def test_dispatches_on_method_name(self, mock_httpx): From 202baee55da6d9c647cb793b8d9bee0fa95cc737 Mon Sep 17 00:00:00 2001 From: Justin Date: Tue, 27 Oct 2026 11:12:12 -0400 Subject: [PATCH 07/13] fix(sentinel): count queue time from submission and read the last definition - start the 15-minute queue deadline when the job is submitted, so time spent waiting on runsync counts toward it - curl example: use the last definition of a redefined function, as python does --- src/runpod_flash/cli/commands/deploy.py | 47 +++++++++++++------------ src/runpod_flash/flash_sentinel.py | 9 +++-- tests/unit/cli/test_deploy.py | 23 ++++++++++++ tests/unit/test_flash_sentinel.py | 26 ++++++++++++++ 4 files changed, 80 insertions(+), 25 deletions(-) diff --git a/src/runpod_flash/cli/commands/deploy.py b/src/runpod_flash/cli/commands/deploy.py index 1faee74c..30b224de 100644 --- a/src/runpod_flash/cli/commands/deploy.py +++ b/src/runpod_flash/cli/commands/deploy.py @@ -148,28 +148,31 @@ def _required_params( return None body = cls.body - for node in body: - if ( - isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) - and node.name == func_name - ): - args = node.args - positional = args.posonlyargs + args.args - required = positional[: len(positional) - len(args.defaults)] - required += [ - a - for a, default in zip(args.kwonlyargs, args.kw_defaults) - if default is None - ] - # a method's receiver is bound by the handler, not sent as input - if class_name is not None and required and positional: - if required[0] is positional[0] and positional[0].arg in ( - "self", - "cls", - ): - required = required[1:] - return [a.arg for a in required] - return None + # the last definition wins at import time, as in python itself + matches = [ + node + for node in body + if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) + and node.name == func_name + ] + if not matches: + return None + + args = matches[-1].args + positional = args.posonlyargs + args.args + required = positional[: len(positional) - len(args.defaults)] + required += [ + a for a, default in zip(args.kwonlyargs, args.kw_defaults) if default is None + ] + # a method's receiver is bound by the handler, not sent as input + if ( + class_name is not None + and required + and required[0] is positional[0] + and positional[0].arg in ("self", "cls") + ): + required = required[1:] + return [a.arg for a in required] def _qb_example_input(resource: dict[str, Any]) -> dict[str, Any]: diff --git a/src/runpod_flash/flash_sentinel.py b/src/runpod_flash/flash_sentinel.py index 89a76b93..59f9b4f8 100644 --- a/src/runpod_flash/flash_sentinel.py +++ b/src/runpod_flash/flash_sentinel.py @@ -131,6 +131,8 @@ async def _sentinel_qb_post( log.debug("sentinel QB -> %s/%s/%s", app, env, endpoint_name) + submitted_at = time.monotonic() + try: async with _http.get_authenticated_httpx_client( timeout=effective_timeout @@ -166,7 +168,7 @@ async def _sentinel_qb_post( f"{data['status']} without a job id: {data}" ) data = await _poll_until_done( - app, env, endpoint_name, data["id"], effective_timeout + app, env, endpoint_name, data["id"], effective_timeout, submitted_at ) return data @@ -177,19 +179,20 @@ async def _poll_until_done( endpoint_name: str, job_id: str, request_timeout: float, + submitted_at: float, ) -> Dict[str, Any]: """poll the sentinel /status route until the job leaves IN_QUEUE/IN_PROGRESS. a running job is bounded by the endpoint's execution timeout, which ends it server-side with TIMED_OUT. a job no worker picks up is cancelled - after _QUEUE_WAIT_MAX_SECONDS. + after _QUEUE_WAIT_MAX_SECONDS, counted from submission. """ base = f"{runpod.endpoint_url_base}/{FLASH_SENTINEL_ID}" headers = _flash_headers(app, env, endpoint_name) log.debug("sentinel QB job %s still pending, polling /status", job_id) - queued_since = time.monotonic() + queued_since = submitted_at attempt = 0 transport_errors = 0 async with _http.get_authenticated_httpx_client(timeout=request_timeout) as client: diff --git a/tests/unit/cli/test_deploy.py b/tests/unit/cli/test_deploy.py index 3b27d779..8c4a36b4 100644 --- a/tests/unit/cli/test_deploy.py +++ b/tests/unit/cli/test_deploy.py @@ -957,6 +957,29 @@ def test_qb_curl_keeps_self_param_on_plain_function( output = self._collect_output(patched_console) assert '{"input": {"self": ""}}' in output + def test_qb_curl_uses_last_definition_of_redefined_function( + self, patched_console, tmp_path + ): + from runpod_flash.cli.commands.deploy import _display_post_deployment_guidance + + src = tmp_path / "jobs.py" + src.write_text( + "def predict(image):\n ...\n\ndef predict(image, scale):\n ...\n" + ) + _display_post_deployment_guidance( + env_name="production", + resources_endpoints={"my_qb": "https://api.runpod.ai/v2/abc123"}, + resources={ + "my_qb": { + "file_path": str(src), + "functions": [{"name": "predict", "is_class": False}], + } + }, + routes={}, + ) + output = self._collect_output(patched_console) + assert '{"input": {"image": "", "scale": ""}}' in output + def test_lb_endpoint_shows_routes(self, patched_console): from runpod_flash.cli.commands.deploy import _display_post_deployment_guidance diff --git a/tests/unit/test_flash_sentinel.py b/tests/unit/test_flash_sentinel.py index 01c46803..63986612 100644 --- a/tests/unit/test_flash_sentinel.py +++ b/tests/unit/test_flash_sentinel.py @@ -445,6 +445,32 @@ async def my_func(): with pytest.raises(RuntimeError, match="cancelling it failed"): await sentinel_qb_execute("myapp", "prod", "gpu-worker", my_func) + @pytest.mark.asyncio + async def test_queue_deadline_counts_time_spent_in_runsync(self, mock_httpx): + from runpod_flash.flash_sentinel import ( + _QUEUE_WAIT_MAX_SECONDS, + sentinel_qb_execute, + ) + + queued = _json_response({"id": "job-1", "status": "IN_QUEUE"}) + mock_client = _make_mock_client(queued) + mock_client.get = AsyncMock(return_value=queued) + mock_httpx.return_value = mock_client + + # submitted at 0, runsync returns pending, first poll is already late + clock = iter([0.0, _QUEUE_WAIT_MAX_SECONDS + 1]) + + async def my_func(): + pass + + with patch( + "runpod_flash.flash_sentinel.time.monotonic", + side_effect=lambda: next(clock), + ): + with pytest.raises(RuntimeError, match="still queued"): + await sentinel_qb_execute("myapp", "prod", "gpu-worker", my_func) + mock_client.get.assert_awaited_once() + @pytest.mark.asyncio async def test_running_job_is_not_cancelled_by_queue_deadline(self, mock_httpx): from runpod_flash.flash_sentinel import ( From d6efc3e52690ab46ca36aa48213a7eb7cc451ada Mon Sep 17 00:00:00 2001 From: Justin Date: Tue, 27 Oct 2026 11:14:54 -0400 Subject: [PATCH 08/13] fix(deploy): read the last definition of a redefined class for the curl example --- src/runpod_flash/cli/commands/deploy.py | 15 +++++---------- tests/unit/cli/test_deploy.py | 19 +++++++++++++++++++ 2 files changed, 24 insertions(+), 10 deletions(-) diff --git a/src/runpod_flash/cli/commands/deploy.py b/src/runpod_flash/cli/commands/deploy.py index 30b224de..d41516be 100644 --- a/src/runpod_flash/cli/commands/deploy.py +++ b/src/runpod_flash/cli/commands/deploy.py @@ -136,17 +136,12 @@ def _required_params( body: list[ast.stmt] = tree.body if class_name is not None: - cls = next( - ( - n - for n in tree.body - if isinstance(n, ast.ClassDef) and n.name == class_name - ), - None, - ) - if cls is None: + classes = [ + n for n in tree.body if isinstance(n, ast.ClassDef) and n.name == class_name + ] + if not classes: return None - body = cls.body + body = classes[-1].body # the last definition wins at import time, as in python itself matches = [ diff --git a/tests/unit/cli/test_deploy.py b/tests/unit/cli/test_deploy.py index 8c4a36b4..d3211e61 100644 --- a/tests/unit/cli/test_deploy.py +++ b/tests/unit/cli/test_deploy.py @@ -912,6 +912,25 @@ def test_qb_curl_for_multi_method_class_names_real_method( output = self._collect_output(patched_console) assert '{"input": {"method": "predict", "image": ""}}' in output + def test_qb_curl_uses_last_definition_of_redefined_class( + self, patched_console, tmp_path + ): + from runpod_flash.cli.commands.deploy import _display_post_deployment_guidance + + src = tmp_path / "model.py" + src.write_text( + "class Model:\n def predict(self, text):\n ...\n\n" + "class Model:\n def predict(self, image):\n ...\n" + ) + _display_post_deployment_guidance( + env_name="production", + resources_endpoints={"my_qb": "https://api.runpod.ai/v2/abc123"}, + resources=self._class_resource(src, ["predict"]), + routes={}, + ) + output = self._collect_output(patched_console) + assert '{"input": {"image": ""}}' in output + def test_qb_curl_reads_top_level_function_not_nested_namesake( self, patched_console, tmp_path ): From 607ab34eecdfa856482719ef600c10a4d2282b28 Mon Sep 17 00:00:00 2001 From: Justin Date: Tue, 27 Oct 2026 12:07:41 -0400 Subject: [PATCH 09/13] fix(sentinel): cap the poll backoff exponent and drop any method receiver - long jobs overflowed float in get_backoff_delay after ~1024 polls (about 85 minutes); keep the attempt count at the cap - curl example: drop an instance method's first parameter whatever it's named, and keep it for static methods --- src/runpod_flash/cli/commands/deploy.py | 19 ++++++++-------- src/runpod_flash/flash_sentinel.py | 4 +++- tests/unit/cli/test_deploy.py | 29 +++++++++++++++++++++++++ tests/unit/test_flash_sentinel.py | 21 ++++++++++++++++++ 4 files changed, 63 insertions(+), 10 deletions(-) diff --git a/src/runpod_flash/cli/commands/deploy.py b/src/runpod_flash/cli/commands/deploy.py index d41516be..ecf0df01 100644 --- a/src/runpod_flash/cli/commands/deploy.py +++ b/src/runpod_flash/cli/commands/deploy.py @@ -153,20 +153,21 @@ def _required_params( if not matches: return None - args = matches[-1].args + func = matches[-1] + args = func.args positional = args.posonlyargs + args.args required = positional[: len(positional) - len(args.defaults)] required += [ a for a, default in zip(args.kwonlyargs, args.kw_defaults) if default is None ] - # a method's receiver is bound by the handler, not sent as input - if ( - class_name is not None - and required - and required[0] is positional[0] - and positional[0].arg in ("self", "cls") - ): - required = required[1:] + # a method's receiver (whatever it's named) is bound by the handler, not + # sent as input. static methods have no receiver. + is_static = any( + isinstance(d, ast.Name) and d.id == "staticmethod" for d in func.decorator_list + ) + if class_name is not None and not is_static and required: + if required[0] is positional[0]: + required = required[1:] return [a.arg for a in required] diff --git a/src/runpod_flash/flash_sentinel.py b/src/runpod_flash/flash_sentinel.py index 59f9b4f8..ed6c5b6c 100644 --- a/src/runpod_flash/flash_sentinel.py +++ b/src/runpod_flash/flash_sentinel.py @@ -202,7 +202,9 @@ async def _poll_until_done( attempt, base=1.0, max_seconds=_STATUS_POLL_MAX_SECONDS ) ) - attempt += 1 + # the delay is already at its cap by now; a growing exponent + # overflows float after ~1000 polls on long jobs + attempt = min(attempt + 1, 8) try: response = await client.get(f"{base}/status/{job_id}", headers=headers) if response.status_code == 429 or response.status_code >= 500: diff --git a/tests/unit/cli/test_deploy.py b/tests/unit/cli/test_deploy.py index d3211e61..8556b5c2 100644 --- a/tests/unit/cli/test_deploy.py +++ b/tests/unit/cli/test_deploy.py @@ -912,6 +912,35 @@ def test_qb_curl_for_multi_method_class_names_real_method( output = self._collect_output(patched_console) assert '{"input": {"method": "predict", "image": ""}}' in output + @pytest.mark.parametrize( + ("source", "expected"), + [ + ( + "class Model:\n def predict(this, image):\n ...\n", + '{"input": {"image": ""}}', + ), + ( + "class Model:\n @staticmethod\n" + " def predict(image):\n ...\n", + '{"input": {"image": ""}}', + ), + ], + ) + def test_qb_curl_drops_only_the_method_receiver( + self, patched_console, tmp_path, source, expected + ): + from runpod_flash.cli.commands.deploy import _display_post_deployment_guidance + + src = tmp_path / "model.py" + src.write_text(source) + _display_post_deployment_guidance( + env_name="production", + resources_endpoints={"my_qb": "https://api.runpod.ai/v2/abc123"}, + resources=self._class_resource(src, ["predict"]), + routes={}, + ) + assert expected in self._collect_output(patched_console) + def test_qb_curl_uses_last_definition_of_redefined_class( self, patched_console, tmp_path ): diff --git a/tests/unit/test_flash_sentinel.py b/tests/unit/test_flash_sentinel.py index 63986612..b82cd700 100644 --- a/tests/unit/test_flash_sentinel.py +++ b/tests/unit/test_flash_sentinel.py @@ -471,6 +471,27 @@ async def my_func(): await sentinel_qb_execute("myapp", "prod", "gpu-worker", my_func) mock_client.get.assert_awaited_once() + @pytest.mark.asyncio + async def test_long_running_job_polls_past_backoff_overflow(self, mock_httpx): + """an uncapped exponent overflows float after ~1024 polls.""" + from runpod_flash.flash_sentinel import sentinel_qb_execute + + running = _json_response({"id": "job-1", "status": "IN_PROGRESS"}) + done = _json_response( + {"id": "job-1", "status": "COMPLETED", "output": {"ok": True}} + ) + mock_client = _make_mock_client(running) + mock_client.get = AsyncMock(side_effect=[running] * 1100 + [done]) + mock_httpx.return_value = mock_client + + async def my_func(): + pass + + result = await sentinel_qb_execute("myapp", "prod", "gpu-worker", my_func) + + assert result == {"ok": True} + assert mock_client.get.call_count == 1101 + @pytest.mark.asyncio async def test_running_job_is_not_cancelled_by_queue_deadline(self, mock_httpx): from runpod_flash.flash_sentinel import ( From 14d6982f476aab98862e19aaf1d730ad6564258d Mon Sep 17 00:00:00 2001 From: Justin Date: Tue, 27 Oct 2026 12:12:20 -0400 Subject: [PATCH 10/13] fix(deploy): recognize qualified staticmethod decorators in the curl example --- src/runpod_flash/cli/commands/deploy.py | 4 +++- tests/unit/cli/test_deploy.py | 5 +++++ 2 files changed, 8 insertions(+), 1 deletion(-) diff --git a/src/runpod_flash/cli/commands/deploy.py b/src/runpod_flash/cli/commands/deploy.py index ecf0df01..5b4b7469 100644 --- a/src/runpod_flash/cli/commands/deploy.py +++ b/src/runpod_flash/cli/commands/deploy.py @@ -163,7 +163,9 @@ def _required_params( # a method's receiver (whatever it's named) is bound by the handler, not # sent as input. static methods have no receiver. is_static = any( - isinstance(d, ast.Name) and d.id == "staticmethod" for d in func.decorator_list + (isinstance(d, ast.Name) and d.id == "staticmethod") + or (isinstance(d, ast.Attribute) and d.attr == "staticmethod") + for d in func.decorator_list ) if class_name is not None and not is_static and required: if required[0] is positional[0]: diff --git a/tests/unit/cli/test_deploy.py b/tests/unit/cli/test_deploy.py index 8556b5c2..ecbd37e1 100644 --- a/tests/unit/cli/test_deploy.py +++ b/tests/unit/cli/test_deploy.py @@ -924,6 +924,11 @@ def test_qb_curl_for_multi_method_class_names_real_method( " def predict(image):\n ...\n", '{"input": {"image": ""}}', ), + ( + "import builtins\n\nclass Model:\n @builtins.staticmethod\n" + " def predict(image):\n ...\n", + '{"input": {"image": ""}}', + ), ], ) def test_qb_curl_drops_only_the_method_receiver( From 715d876603b5b8821102feccdef6c36885d0fed2 Mon Sep 17 00:00:00 2001 From: Justin Date: Tue, 27 Oct 2026 13:00:25 -0400 Subject: [PATCH 11/13] fix(sentinel): raise a RuntimeError naming the job on a 4xx while polling --- src/runpod_flash/flash_sentinel.py | 7 ++++++- tests/unit/test_flash_sentinel.py | 3 ++- 2 files changed, 8 insertions(+), 2 deletions(-) diff --git a/src/runpod_flash/flash_sentinel.py b/src/runpod_flash/flash_sentinel.py index ed6c5b6c..bb4b8cbd 100644 --- a/src/runpod_flash/flash_sentinel.py +++ b/src/runpod_flash/flash_sentinel.py @@ -219,7 +219,12 @@ async def _poll_until_done( log.debug("polling job %s failed (%s), retrying", job_id, exc) continue transport_errors = 0 - response.raise_for_status() + try: + response.raise_for_status() + except httpx.HTTPStatusError as exc: + raise RuntimeError( + f"checking job {job_id} on endpoint '{endpoint_name}' failed: {exc}" + ) from exc data: Dict[str, Any] = response.json() status = data.get("status") if status not in _PENDING_STATUSES: diff --git a/tests/unit/test_flash_sentinel.py b/tests/unit/test_flash_sentinel.py index b82cd700..4495f808 100644 --- a/tests/unit/test_flash_sentinel.py +++ b/tests/unit/test_flash_sentinel.py @@ -362,8 +362,9 @@ async def test_poll_client_error_propagates(self, mock_httpx): async def my_func(): pass - with pytest.raises(httpx.HTTPStatusError): + with pytest.raises(RuntimeError, match="job job-1.*gpu-worker") as exc_info: await sentinel_qb_execute("myapp", "prod", "gpu-worker", my_func) + assert isinstance(exc_info.value.__cause__, httpx.HTTPStatusError) mock_client.get.assert_awaited_once() @pytest.mark.asyncio From 9da906659b73b4ae3c9528fa94a8640dc5df610b Mon Sep 17 00:00:00 2001 From: Justin Date: Tue, 27 Oct 2026 14:16:09 -0400 Subject: [PATCH 12/13] fix(sentinel): address review: document unbounded polling, wrap non-JSON status, tie backoff cap to max delay --- src/runpod_flash/cli/docs/README.md | 4 ++- src/runpod_flash/flash_sentinel.py | 44 +++++++++++++++++++++-------- tests/unit/test_flash_sentinel.py | 32 +++++++++++++++++++++ 3 files changed, 67 insertions(+), 13 deletions(-) diff --git a/src/runpod_flash/cli/docs/README.md b/src/runpod_flash/cli/docs/README.md index 0feca0e7..39f410cb 100644 --- a/src/runpod_flash/cli/docs/README.md +++ b/src/runpod_flash/cli/docs/README.md @@ -281,9 +281,11 @@ Optional: ```bash FLASH_APP=my-project # defaults to current directory name FLASH_ENV=staging # defaults to "production" -FLASH_SENTINEL_TIMEOUT=120 # sentinel request timeout in seconds (default: 90) +FLASH_SENTINEL_TIMEOUT=120 # timeout for each sentinel HTTP request in seconds (default: 90) ``` +`FLASH_SENTINEL_TIMEOUT` doesn't cap how long a call waits. A job that's still queued or running is polled until it finishes. A job no worker picks up within 15 minutes is cancelled. To bound a running job, set `execution_timeout_ms` on the `Endpoint`. The default of 0 sets no limit. + ## Testing your dev server ```bash diff --git a/src/runpod_flash/flash_sentinel.py b/src/runpod_flash/flash_sentinel.py index bb4b8cbd..8f54e946 100644 --- a/src/runpod_flash/flash_sentinel.py +++ b/src/runpod_flash/flash_sentinel.py @@ -14,6 +14,7 @@ import base64 import inspect import logging +import math import os import time from typing import Any, Callable, Dict, Optional @@ -31,8 +32,9 @@ FLASH_SENTINEL_ID = "flash" -# default timeout for sentinel requests. configurable via -# FLASH_SENTINEL_TIMEOUT env var (seconds). +# default timeout for each sentinel HTTP request. configurable via +# FLASH_SENTINEL_TIMEOUT env var (seconds). it doesn't cap the total wait: +# a pending job is polled until it finishes. DEFAULT_SENTINEL_TIMEOUT = 90 # runsync returns one of these when the job outlives the server-side wait @@ -42,11 +44,22 @@ # longest gap between /status polls, in seconds. _STATUS_POLL_MAX_SECONDS = 5.0 +# backoff exponent at which the poll delay reaches _STATUS_POLL_MAX_SECONDS +# (with a 1s base). capping the exponent there keeps 2**attempt from +# overflowing float after ~1000 polls on long jobs. +_STATUS_POLL_BASE_SECONDS = 1.0 +_STATUS_POLL_MAX_ATTEMPT = math.ceil( + math.log2(_STATUS_POLL_MAX_SECONDS / _STATUS_POLL_BASE_SECONDS) +) + # give up (and cancel) when no worker picks the job up for this long. -# once the job is IN_PROGRESS, the endpoint's execution timeout bounds it. +# once the job is IN_PROGRESS, only the endpoint's execution_timeout_ms +# bounds it. flash defaults that to 0 (no limit), so a hung handler is +# polled until the job's 24h TTL expires. _QUEUE_WAIT_MAX_SECONDS = 900.0 -# consecutive network errors or 429/5xx responses tolerated while polling. +# polling raises after this many consecutive network errors or 429/5xx +# responses. _STATUS_POLL_MAX_TRANSPORT_ERRORS = 5 # runsync holds the request for 90s by default, the same as the default @@ -183,9 +196,10 @@ async def _poll_until_done( ) -> Dict[str, Any]: """poll the sentinel /status route until the job leaves IN_QUEUE/IN_PROGRESS. - a running job is bounded by the endpoint's execution timeout, which ends - it server-side with TIMED_OUT. a job no worker picks up is cancelled - after _QUEUE_WAIT_MAX_SECONDS, counted from submission. + a running job is bounded only by the endpoint's execution_timeout_ms, + which ends it server-side with TIMED_OUT. the default of 0 sets no limit. + a job no worker picks up is cancelled after _QUEUE_WAIT_MAX_SECONDS, + counted from submission. """ base = f"{runpod.endpoint_url_base}/{FLASH_SENTINEL_ID}" headers = _flash_headers(app, env, endpoint_name) @@ -199,12 +213,12 @@ async def _poll_until_done( while True: await asyncio.sleep( get_backoff_delay( - attempt, base=1.0, max_seconds=_STATUS_POLL_MAX_SECONDS + attempt, + base=_STATUS_POLL_BASE_SECONDS, + max_seconds=_STATUS_POLL_MAX_SECONDS, ) ) - # the delay is already at its cap by now; a growing exponent - # overflows float after ~1000 polls on long jobs - attempt = min(attempt + 1, 8) + attempt = min(attempt + 1, _STATUS_POLL_MAX_ATTEMPT) try: response = await client.get(f"{base}/status/{job_id}", headers=headers) if response.status_code == 429 or response.status_code >= 500: @@ -225,7 +239,13 @@ async def _poll_until_done( raise RuntimeError( f"checking job {job_id} on endpoint '{endpoint_name}' failed: {exc}" ) from exc - data: Dict[str, Any] = response.json() + try: + data: Dict[str, Any] = response.json() + except ValueError as exc: + raise RuntimeError( + f"checking job {job_id} on endpoint '{endpoint_name}' " + f"returned a non-JSON response: {response.text[:200]!r}" + ) from exc status = data.get("status") if status not in _PENDING_STATUSES: return data diff --git a/tests/unit/test_flash_sentinel.py b/tests/unit/test_flash_sentinel.py index 4495f808..a9520d16 100644 --- a/tests/unit/test_flash_sentinel.py +++ b/tests/unit/test_flash_sentinel.py @@ -367,6 +367,38 @@ async def my_func(): assert isinstance(exc_info.value.__cause__, httpx.HTTPStatusError) mock_client.get.assert_awaited_once() + @pytest.mark.asyncio + async def test_poll_non_json_response_raises(self, mock_httpx): + import json + + from runpod_flash.flash_sentinel import sentinel_qb_execute + + html_response = _json_response(None) + html_response.json.side_effect = json.JSONDecodeError("Expecting value", "", 0) + html_response.text = "bad gateway" + mock_client = _make_mock_client( + _json_response({"id": "job-1", "status": "IN_PROGRESS"}) + ) + mock_client.get = AsyncMock(return_value=html_response) + mock_httpx.return_value = mock_client + + async def my_func(): + pass + + with pytest.raises(RuntimeError, match="job job-1.*gpu-worker.*non-JSON"): + await sentinel_qb_execute("myapp", "prod", "gpu-worker", my_func) + + def test_poll_attempt_cap_reaches_max_delay(self): + from runpod_flash.flash_sentinel import ( + _STATUS_POLL_BASE_SECONDS, + _STATUS_POLL_MAX_ATTEMPT, + _STATUS_POLL_MAX_SECONDS, + ) + + capped = _STATUS_POLL_BASE_SECONDS * 2**_STATUS_POLL_MAX_ATTEMPT + below = _STATUS_POLL_BASE_SECONDS * 2 ** (_STATUS_POLL_MAX_ATTEMPT - 1) + assert capped >= _STATUS_POLL_MAX_SECONDS > below + @pytest.mark.asyncio @pytest.mark.parametrize("status_code", [429, 503]) async def test_poll_recovers_from_transient_status(self, mock_httpx, status_code): From 5bbffa06a14605643b485218ad1cf137d9bd1cfe Mon Sep 17 00:00:00 2001 From: Justin Date: Tue, 27 Oct 2026 14:29:48 -0400 Subject: [PATCH 13/13] fix(cli): correct --python-version help and python_version field description --- src/runpod_flash/cli/commands/build.py | 6 +++--- src/runpod_flash/cli/commands/deploy.py | 5 +++-- src/runpod_flash/core/resources/serverless.py | 2 +- 3 files changed, 7 insertions(+), 6 deletions(-) diff --git a/src/runpod_flash/cli/commands/build.py b/src/runpod_flash/cli/commands/build.py index 23e99506..5a29c057 100644 --- a/src/runpod_flash/cli/commands/build.py +++ b/src/runpod_flash/cli/commands/build.py @@ -564,9 +564,9 @@ def build_command( None, "--python-version", help=( - "Target Python version for worker images (3.10, 3.11, or 3.12). " - "Overrides per-resource python_version declarations. " - "Defaults to the version declared on resource configs, or 3.12 if none set." + "Target Python version for worker images (3.10, 3.11, 3.12, or 3.13). " + "Defaults to the version declared on resource configs, or 3.12 if none " + "set. Fails if it conflicts with a declared python_version." ), ), ): diff --git a/src/runpod_flash/cli/commands/deploy.py b/src/runpod_flash/cli/commands/deploy.py index 5b4b7469..870491a1 100644 --- a/src/runpod_flash/cli/commands/deploy.py +++ b/src/runpod_flash/cli/commands/deploy.py @@ -51,8 +51,9 @@ def deploy_command( None, "--python-version", help=( - "Target Python version for worker images (3.10, 3.11, or 3.12). " - "Overrides per-resource python_version declarations." + "Target Python version for worker images (3.10, 3.11, 3.12, or 3.13). " + "Defaults to the version declared on resource configs, or 3.12 if none " + "set. Fails if it conflicts with a declared python_version." ), ), ): diff --git a/src/runpod_flash/core/resources/serverless.py b/src/runpod_flash/core/resources/serverless.py index 67f6eb5a..5f4af5c5 100644 --- a/src/runpod_flash/core/resources/serverless.py +++ b/src/runpod_flash/core/resources/serverless.py @@ -276,7 +276,7 @@ class ServerlessResource(DeployableResource): datacenter: Optional[List[DataCenter] | DataCenter] = Field(default=None) python_version: Optional[str] = Field( default=None, - description="Python version for runtime image selection. Defaults to the local interpreter version at build time.", + description="Python version for runtime image selection. Defaults to 3.12 when no resource declares one.", ) # === Input Fields ===