feat(serverless): report fitness failure to host before force-exit

runpod/serverless/modules/rp_fitness.py

@@ -110,6 +110,51 @@ def _reset_registration_state() -> None:
     _registration_state["system_checks"] = False
 
 
+# Bound how long the best-effort unhealthy report may delay the exit.
+_REPORT_TIMEOUT_SECONDS = 2
+
+
+def _report_unhealthy(check: str, reason: str) -> None:
+    """
+    Best-effort report of a fitness-check failure to the host before exit.
+
+    Sends a single GET to the ping URL (same URL/credentials the heartbeat
+    uses) with status=unhealthy plus the failing check name and reason, so the
+    host can emit a queryable worker.fitness_failed event. Any failure — no
+    ping URL, no API key, HTTP error, timeout — is swallowed, so this can never
+    prevent the os._exit that follows. It is synchronous, so it may delay that
+    exit by up to _REPORT_TIMEOUT_SECONDS (network phases only; it adds no
+    delay when there is no ping URL/API key to report to).
+    """
+    ping_url = os.environ.get("RUNPOD_WEBHOOK_PING")
+    api_key = os.environ.get("RUNPOD_AI_API_KEY")
+    if not ping_url or ping_url == "PING_NOT_SET" or not api_key:
+        return
+
+    try:
+        # Deferred imports: keep module import light and avoid import cycles.
+        from runpod.http_client import SyncClientSession
+        from runpod.serverless.modules.worker_state import WORKER_ID
+        from runpod.version import __version__ as runpod_version
+
+        ping_url = ping_url.replace("$RUNPOD_POD_ID", WORKER_ID)
+        params = {
+            "status": "unhealthy",
+            "check": check,
+            "reason": reason[:256],
+            "runpod_version": runpod_version,
+        }
+        session = SyncClientSession()
+        try:
+            session.headers.update({"Authorization": api_key})
+            session.get(ping_url, params=params, timeout=_REPORT_TIMEOUT_SECONDS)
+        finally:
+            session.close()
+    except Exception:
+        # Best-effort only; the exit is the guarantee, not this report.
+        pass
+
+
 def _ensure_gpu_check_registered() -> None:
     """
     Ensure GPU fitness check is registered.
@@ -234,6 +279,14 @@ async def run_fitness_checks() -> None:
             )
             log.debug(f"Traceback:\n{full_traceback}")
 
+            # Best-effort report to the host so the failure is queryable. It is
+            # bounded (see _REPORT_TIMEOUT_SECONDS) and fully swallowed, so it
+            # can delay the force-exit below but can never prevent it.
+            try:
+                _report_unhealthy(check_name, f"{error_type}: {error_message}")
+            except Exception:  # a report failure must never prevent the exit
+                pass
+
             # Force-kill immediately; see _terminate_unhealthy for why this is
             # os._exit rather than sys.exit.
             log.error("Worker is unhealthy, exiting.")

tests/test_serverless/test_modules/test_fitness/test_force_kill.py

@@ -7,11 +7,16 @@
 raises ``SystemExit`` and then blocks in interpreter shutdown joining those
 threads, so the worker logs "unhealthy, exiting." but never terminates.
 The exit must go through ``os._exit`` to terminate unconditionally.
+
+Also covers SLS-380: the best-effort unhealthy report the worker sends to the
+host before force-exiting.
 """
 
 import subprocess
 import sys
-from unittest.mock import patch
+from unittest.mock import MagicMock, patch
+
+import pytest
 
 from runpod.serverless.modules import rp_fitness
 
@@ -92,3 +97,78 @@ def test_terminate_unhealthy_exits_even_if_flush_raises():
         rp_fitness._terminate_unhealthy(1)
 
     mock_exit.assert_called_once_with(1)
+
+
+def test_report_unhealthy_posts_check_and_reason(monkeypatch):
+    monkeypatch.setenv("RUNPOD_WEBHOOK_PING", "https://api.test/ping/$RUNPOD_POD_ID")
+    monkeypatch.setenv("RUNPOD_AI_API_KEY", "key-123")
+
+    fake_session = MagicMock()
+    with patch("runpod.http_client.SyncClientSession", return_value=fake_session), \
+        patch("runpod.serverless.modules.worker_state.WORKER_ID", "podABC"):
+        rp_fitness._report_unhealthy("_cuda_init_check", "RuntimeError: boom")
+
+    assert fake_session.get.call_count == 1
+    call = fake_session.get.call_args
+    url = call.args[0] if call.args else call.kwargs["url"]
+    params = call.kwargs["params"]
+    assert "podABC" in url  # $RUNPOD_POD_ID substituted
+    assert params["status"] == "unhealthy"
+    assert params["check"] == "_cuda_init_check"
+    assert params["reason"] == "RuntimeError: boom"
+    fake_session.headers.update.assert_called_once_with({"Authorization": "key-123"})
+
+
+def test_report_unhealthy_skipped_without_ping_url(monkeypatch):
+    monkeypatch.delenv("RUNPOD_WEBHOOK_PING", raising=False)
+    monkeypatch.setenv("RUNPOD_AI_API_KEY", "key-123")
+    with patch("runpod.http_client.SyncClientSession") as session_cls:
+        rp_fitness._report_unhealthy("_memory_check", "RuntimeError: low")
+    session_cls.assert_not_called()
+
+
+def test_report_unhealthy_truncates_long_reason(monkeypatch):
+    monkeypatch.setenv("RUNPOD_WEBHOOK_PING", "https://api.test/ping")
+    monkeypatch.setenv("RUNPOD_AI_API_KEY", "key-123")
+
+    fake_session = MagicMock()
+    with patch("runpod.http_client.SyncClientSession", return_value=fake_session):
+        rp_fitness._report_unhealthy("_disk_check", "x" * 300)
+
+    params = fake_session.get.call_args.kwargs["params"]
+    assert len(params["reason"]) == 256
+
+
+def test_report_unhealthy_skipped_without_api_key(monkeypatch):
+    monkeypatch.setenv("RUNPOD_WEBHOOK_PING", "https://api.test/ping")
+    monkeypatch.delenv("RUNPOD_AI_API_KEY", raising=False)
+    with patch("runpod.http_client.SyncClientSession") as session_cls:
+        rp_fitness._report_unhealthy("_memory_check", "RuntimeError: low")
+    session_cls.assert_not_called()
+
+
+def test_report_unhealthy_swallows_errors(monkeypatch):
+    monkeypatch.setenv("RUNPOD_WEBHOOK_PING", "https://api.test/ping")
+    monkeypatch.setenv("RUNPOD_AI_API_KEY", "key-123")
+    fake_session = MagicMock()
+    fake_session.get.side_effect = RuntimeError("network down")
+    with patch("runpod.http_client.SyncClientSession", return_value=fake_session):
+        rp_fitness._report_unhealthy("_disk_check", "RuntimeError: full")  # must not raise
+
+
+@pytest.mark.asyncio
+async def test_failure_reports_before_exit():
+    from runpod.serverless.modules.rp_fitness import register_fitness_check, run_fitness_checks
+
+    @register_fitness_check
+    def _cuda_init_check():
+        raise RuntimeError("device busy")
+
+    with patch("runpod.serverless.modules.rp_fitness._report_unhealthy") as mock_report:
+        with pytest.raises(SystemExit):
+            await run_fitness_checks()
+
+    mock_report.assert_called_once()
+    args = mock_report.call_args.args
+    assert args[0] == "_cuda_init_check"
+    assert "RuntimeError" in args[1] and "device busy" in args[1]